# 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/siv2r/bip-frost-signing, `README.md` at v0.12.1,
head `ba299badfbe647e670686de75578dd501a627028`. The BIP is a draft,
not in `bitcoin/bips` `master`; `tests/_data/README.md` records why
this module pins that repository rather than its pull request,
https://github.com/bitcoin/bips/pull/2070.
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 `ecc.ssa` and with any other BIP340
verifier -- the same output shape `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. It is tested
on BIP445's vectors, which a trusted dealer generated, and on the output
of ChillDKG's reference. `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#2200` is
where key generation is tracked: a trusted dealer, then ChillDKG, once
BIP445 and BIP-FROST-DKG settle.
The round shape mirrors `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_var` and
`apply_tweak_var`, or `thresh_pubkey_and_tweak_var` for both in one call,
answering a `TweakContext`;
- round 1, the nonces: `nonce_gen` for each of the *u* selected signers,
then `nonce_agg_var` 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 in `dsa` and `ssa`,
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 `ecc.ssa.challenge_`.
**Nothing here is shared with `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_var`, `apply_tweak_var` -- 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 ecc import ssa
from ecc._utils import assert_type, bytes_from_octets, is_integer, is_octets
from ecc.alias import INF, Octets, Point, Public
from ecc.curves import mult, multi_mult_var, secp256k1
from ecc.curves.curve import _sum_var, _tweak_add_var
from ecc.curves.sec_point import bytes_from_point, point_from_octets_var
from ecc.exceptions import (
EccTypeError,
EccValueError,
InvalidContributionError,
)
from ecc.hashes import tagged_hash
from ecc.number_theory import mod_inv_var
__all__ = [
"ID_SIZE",
"MAX_PARTICIPANTS",
"PK_SIZE",
"SessionContext",
"SessionValues",
"ThresholdInfo",
"TweakContext",
"apply_tweak_var",
"deterministic_sign",
"nonce_agg_var",
"nonce_gen",
"nonce_gen_",
"partial_sig_agg",
"partial_sig_verify",
"partial_sig_verify_",
"session_values",
"sign",
"thresh_pubkey_and_tweak_var",
"tweak_ctx_init_var",
"validate_threshold_info",
]
# BIP445's own tags: disjoint from BIP327's, so nothing here collides
# with `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/schemes/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 tweaked threshold public key must not be the point at infinity."
_TWEAKS_LEN_ERR = "The tweaks and is_xonly lists 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."
_NONCE_SIGNER_ID_ERR = f"The optional signer_id must be less than {MAX_PARTICIPANTS}."
_SEC_SHARE_RANGE_ERR = "The secshare is out of range."
_SIGNER_ID_MISSING_ERR = "The signer's id is missing from the ids list."
_AGG_OTHER_NONCE_ABSENT_ERR = "The aggothernonce must be omitted when u = 1."
_AGG_OTHER_NONCE_PRESENT_ERR = (
"The aggothernonce must be present and a 66-byte array when u > 1."
)
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_var(octets: Octets) -> Point:
"""Deserialize a compressed point, refusing anything else."""
return point_from_octets_var(bytes_from_octets(octets, PK_SIZE), secp256k1)
def _cpoint_ext_var(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_var(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 EccTypeError(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 EccTypeError(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 EccValueError(_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_var(ids: Sequence[Public], signer_id: Public) -> int:
"""Return the Lagrange coefficient of `signer_id` for the secret at x = -1.
Assumes `ids` holds no duplicate, which `_validate_session_params`
has checked before either caller reaches this. `signer_id in ids` is
checked by its callers, `sign` and `partial_sig_verify_`: an id the
session does not hold has no `curr_id == signer_id` to match, and the
coefficient returned interpolates nothing.
"""
num = 1
deno = 1
for curr_id in ids:
if curr_id == signer_id:
continue
num = num * (curr_id + 1) % secp256k1.n
deno = deno * (curr_id - signer_id) % secp256k1.n
return num * mod_inv_var(Public(deno), secp256k1.n) % secp256k1.n
def _derive_pubshare_at_var(
ids: Sequence[Public], pub_shares: Sequence[Point], x: Public
) -> Point:
"""Interpolate the polynomial `pub_shares` sit on, evaluated at `x`.
The general form `_derive_interpolating_value_var` 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_var` -- says it must not be.
"""
coefficients: list[Public] = []
for signer_id in ids:
num = 1
deno = 1
for curr_id in ids:
if curr_id == signer_id:
continue
num = num * (x - curr_id) % secp256k1.n
deno = deno * (signer_id - curr_id) % secp256k1.n
coefficients.append(
Public(num * mod_inv_var(Public(deno), 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_var(
ids: Sequence[Public], pub_shares: Sequence[Point]
) -> bytes:
"""Reconstruct the threshold public key `pub_shares` were split from."""
Q = _derive_pubshare_at_var(ids, pub_shares, Public(-1))
if Q[1] == 0:
raise EccValueError(_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#2200` 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_var(
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_var(pub_share)
except EccValueError as e:
raise EccValueError(
f"The pubshare at index {i} is not a valid point."
) 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.
"""
t, thresh_pk, pub_shares = info.t, info.thresh_pk, info.pub_shares
n = len(pub_shares)
if not 1 <= t <= n:
raise EccValueError(_THRESHOLD_RANGE_ERR)
if n > MAX_PARTICIPANTS:
raise EccValueError(_MAX_PARTICIPANTS_ERR)
try:
_cpoint_var(thresh_pk)
except EccValueError as e:
raise EccValueError("The thresh_pk is not a valid point.") from e
parsed_ids, parsed_points = _parse_present_pub_shares_var(pub_shares)
if len(parsed_ids) < t:
raise EccValueError("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_var(
[Public(v) for v in base_ids], base_points, Public(i)
)
!= point
):
raise EccValueError(_KEY_MATERIAL_POLY_ERR)
if (
_derive_thresh_pubkey_var([Public(v) for v in base_ids], base_points)
!= thresh_pk
):
raise EccValueError(_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_var` 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_var(thresh_pk: Octets) -> TweakContext:
"""Return a fresh `TweakContext` over an untweaked threshold public key.
BIP445's TweakCtxInit.
"""
return TweakContext(_cpoint_var(thresh_pk), 1, 0)
[docs]
def apply_tweak_var(
tweak_ctx: TweakContext, tweak: Octets, is_xonly: bool
) -> TweakContext:
"""Return the context tweaked by t, x-only or plain: BIP445's ApplyTweak.
`musig2.apply_tweak_var`'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 EccValueError(_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 EccValueError(_TWEAK_RANGE_ERR)
Q = _tweak_add_var(Q if g == 1 else secp256k1.negate(Q), Public(t), secp256k1)
if Q[1] == 0:
raise EccValueError(_TWEAK_INF_ERR)
return TweakContext(Q, g * gacc % secp256k1.n, (t + g * tacc) % secp256k1.n)
[docs]
def thresh_pubkey_and_tweak_var(
thresh_pk: Octets, tweaks: Sequence[Octets], is_xonly: Sequence[bool]
) -> TweakContext:
"""Build a `TweakContext` and apply the tweaks in order.
BIP445's TweakCtxInit followed by ApplyTweak for each tweak.
"""
_assert_octets_sequence(tweaks, "tweaks")
tweaks_ = _tweaks(tweaks)
is_xonly_ = _flags(is_xonly)
_require_same_length(tweaks_, is_xonly_)
tweak_ctx = tweak_ctx_init_var(thresh_pk)
for tweak, xonly in zip(tweaks_, is_xonly_, strict=True):
tweak_ctx = apply_tweak_var(tweak_ctx, tweak, xonly)
return tweak_ctx
def _nonce_hash(
rand: bytes,
signer_id: bytes,
pub_share: bytes,
thresh_pk_xonly: bytes,
i: int,
msg_prefixed: bytes,
extra_in: bytes,
) -> int:
buf = b"".join(
[
rand,
len(signer_id).to_bytes(1, "big"),
signer_id,
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,
signer_id: int | 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`, `signer_id`, `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 signer_id is None:
signer_id_bytes = b""
else:
signer_id = _integer(signer_id, "signer_id")
if not 0 <= signer_id < MAX_PARTICIPANTS:
raise EccValueError(_NONCE_SIGNER_ID_ERR)
signer_id_bytes = signer_id.to_bytes(ID_SIZE, "big")
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,
signer_id_bytes,
pub_share_bytes,
thresh_pk_xonly_bytes,
0,
msg_prefixed,
extra,
)
% secp256k1.n
)
k_2 = (
_nonce_hash(
rand,
signer_id_bytes,
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,
signer_id: int | 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,
signer_id,
pub_share,
thresh_pk_xonly,
msg,
extra_in,
)
[docs]
def nonce_agg_var(pub_nonces: Sequence[Octets]) -> bytes:
"""Aggregate the public nonces of round 1 into the 66-byte aggnonce.
BIP445's NonceAgg.
`musig2.nonce_agg_var`'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.
An empty list is refused: BIP445 asks for at least one nonce.
"""
_assert_octets_sequence(pub_nonces, "pub_nonces")
if not pub_nonces:
raise EccValueError("The pubnonces list must not be empty.")
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_var(pub_nonce[j * PK_SIZE : (j + 1) * PK_SIZE]))
except EccValueError 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_var(
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_var` 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 EccValueError(f"The id at index {idx} must satisfy 0 <= id <= n - 1.")
if pub_shares is not None:
try:
pub_share_points.append(_cpoint_var(pub_shares[idx]))
except EccValueError as e:
raise EccValueError(
f"The pubshare at index {idx} is not a valid point."
) 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 EccValueError(_THRESHOLD_RANGE_ERR)
if n > MAX_PARTICIPANTS:
raise EccValueError(_MAX_PARTICIPANTS_ERR)
if not t <= len(ids) <= n:
raise EccValueError("The number of signers must satisfy t <= u <= n.")
if pub_shares is not None and len(pub_shares) != len(ids):
raise EccValueError("The pubshares and ids lists must have the same length.")
pub_share_points = _parse_session_pub_shares_var(ids, n, pub_shares)
if _has_duplicates(ids):
raise EccValueError("The ids list contains duplicate elements.")
if (
pub_shares is not None
and _derive_thresh_pubkey_var([Public(v) for v in ids], pub_share_points)
!= thresh_pk
):
raise EccValueError(_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_var`
- 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_var` 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.
A `session_ctx` that is not a `SessionContext` is refused here as
`EccTypeError`, rather than leaking an `AttributeError` from the
first field read (issue btclib-org/ecc#367).
"""
assert_type(session_ctx, SessionContext, "session_ctx")
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_var(
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_var(session_ctx.agg_nonce[:PK_SIZE])
R_2 = _cpoint_ext_var(session_ctx.agg_nonce[PK_SIZE:])
except EccValueError 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,
signer_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.
The lengths of the secnonce and the secshare are checked, and
`session_values` runs, before the nonce is read, for the reason
`musig2.sign`'s own docstring gives: a call refused before the read
leaves the bytearray untouched, so it may be used again.
`sec_nonce` is checked exactly `bytearray` first, for the reason
`musig2.sign`'s own docstring gives (issue
btclib-org/ecc#11).
"""
assert_type(sec_nonce, bytearray, "sec_nonce")
assert_type(session_ctx, SessionContext, "session_ctx")
if len(sec_nonce) != 2 * _SCALAR_SIZE:
raise EccValueError("The secnonce must be a 64-byte array.")
d_bytes = bytes_from_octets(sec_share, _SCALAR_SIZE)
_integer(signer_id, "signer_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 EccValueError("The first half of secnonce is out of range.")
if not 0 < k_2_ < secp256k1.n:
raise EccValueError("The second half of secnonce is out of range.")
if values.R[1] % 2:
k_1_, k_2_ = secp256k1.n - k_1_, secp256k1.n - k_2_
d_ = int.from_bytes(d_bytes, "big")
if not 0 < d_ < secp256k1.n:
raise EccValueError(_SEC_SHARE_RANGE_ERR)
pub_share = _cbytes(mult(d_, ec=secp256k1))
if signer_id not in session_ctx.ids:
raise EccValueError(_SIGNER_ID_MISSING_ERR)
if session_ctx.pub_shares is not None:
j = session_ctx.ids.index(signer_id)
if session_ctx.pub_shares[j] != pub_share:
raise EccValueError(
"The signer's pubshare does not match its entry in the pubshares list."
)
a = _derive_interpolating_value_var(
[Public(v) for v in session_ctx.ids], Public(signer_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(
masked_sec_share: bytes,
signer_id: int,
ids: Sequence[int],
agg_other_nonce: bytes,
tweaked_thresh_pk_xonly: bytes,
msg: bytes,
i: int,
) -> int:
buf = b"".join(
[
masked_sec_share,
signer_id.to_bytes(ID_SIZE, "big"),
len(ids).to_bytes(ID_SIZE, "big"),
_serialize_ids(ids),
len(agg_other_nonce).to_bytes(1, "big"),
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")
def _agg_other_nonce_bytes(agg_other_nonce: Octets | None, u: int) -> bytes:
"""Return `deterministic_sign`'s aggothernonce, empty for a sole signer.
Absent exactly where `u`, the number of signers, is 1, and otherwise
`_NONCE_SIZE` octets.
"""
agg_other = None if agg_other_nonce is None else bytes_from_octets(agg_other_nonce)
if u == 1:
if agg_other is not None:
raise EccValueError(_AGG_OTHER_NONCE_ABSENT_ERR)
elif agg_other is None or len(agg_other) != _NONCE_SIZE:
raise EccValueError(_AGG_OTHER_NONCE_PRESENT_ERR)
return b"" if agg_other is None else agg_other
def _masked_sec_share(
sec_share: Octets, tweak_ctx: TweakContext, aux_rand: Octets | None
) -> bytes:
"""Return the secret share as `sign` uses it, masked by any `aux_rand`.
`deterministic_sign`'s docstring says why the share is the one `sign`
uses rather than the one given.
"""
d_ = int.from_bytes(bytes_from_octets(sec_share, _SCALAR_SIZE), "big")
if not 0 < d_ < secp256k1.n:
raise EccValueError(_SEC_SHARE_RANGE_ERR)
g = 1 if tweak_ctx.Q[1] % 2 == 0 else secp256k1.n - 1
d = (g * tweak_ctx.gacc * d_ % secp256k1.n).to_bytes(_SCALAR_SIZE, "big")
if aux_rand is None:
return d
return _bytes_xor(
d, tagged_hash(_AUX_TAG, bytes_from_octets(aux_rand, _SCALAR_SIZE))
)
[docs]
def deterministic_sign(
sec_share: Octets,
signer_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.
The nonce is derived from the secret share as `sign` uses it, negated
where the tweaked threshold public key has an odd y and multiplied by
`gacc`. Were it derived from the share as given, a coordinator
replaying the session under the key and under its negation, with no
public shares, would get two partial signatures over one nonce, and
their difference is the share.
`agg_other_nonce` is `None` for a sole signer, who has no other
signers' nonces to aggregate, and for nobody else; 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)
signer_id = _integer(signer_id, "signer_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)
if signer_id not in ids_:
raise EccValueError(_SIGNER_ID_MISSING_ERR)
agg_other_nonce_bytes = _agg_other_nonce_bytes(agg_other_nonce, len(ids_))
tweak_ctx = thresh_pubkey_and_tweak_var(thresh_pk_bytes, tweaks, is_xonly)
tweaked_thresh_pk_xonly = tweak_ctx.x_only_pub_key
masked_sec_share = _masked_sec_share(sec_share, tweak_ctx, aux_rand)
msg_bytes = bytes_from_octets(msg)
k_1 = (
_det_nonce_hash(
masked_sec_share,
signer_id,
ids_,
agg_other_nonce_bytes,
tweaked_thresh_pk_xonly,
msg_bytes,
0,
)
% secp256k1.n
)
k_2 = (
_det_nonce_hash(
masked_sec_share,
signer_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_var([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, signer_id, session_ctx)
[docs]
def partial_sig_verify_(
psig: Octets,
signer_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.
"""
assert_type(session_ctx, SessionContext, "session_ctx")
_integer(signer_id, "signer_id")
values = session_values(session_ctx)
if signer_id not in session_ctx.ids:
raise EccValueError(_SIGNER_ID_MISSING_ERR)
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_var(pub_nonce[:PK_SIZE])
R_s2 = _cpoint_var(pub_nonce[PK_SIZE:])
except EccValueError:
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_var(pub_share)
except EccValueError:
return False
if (
session_ctx.pub_shares is not None
and _cbytes(P) != session_ctx.pub_shares[session_ctx.ids.index(signer_id)]
):
return False
a = _derive_interpolating_value_var(
[Public(v) for v in session_ctx.ids], Public(signer_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, and `siv2r/bip-frost-signing`'s
`python/vectors/test_vectors_summary.md` says the same 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 EccValueError(err_msg)
if not 0 <= i < len(ids):
raise EccValueError("The signer index must satisfy 0 <= i <= u - 1.")
agg_nonce = nonce_agg_var(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 `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_type(session_ctx, SessionContext, "session_ctx")
_assert_octets_sequence(psigs, "psigs")
values = session_values(session_ctx)
if len(psigs) != len(session_ctx.ids):
raise EccValueError("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)