feat(taproot): taproot secrets for v3 keysets (NUT-10 POC)

.claude/ralph-loop.local.md

@@ -0,0 +1,10 @@
+---
+active: true
+iteration: 6
+session_id: 1e7a9591-fabc-45e1-807a-beda3e17fb88
+max_iterations: 10
+completion_promise: "TAPROOT V3 ALL MILESTONES GREEN"
+started_at: "2026-07-24T19:54:57Z"
+---
+
+Read taproot-v3/PROMPT.md and execute it: one task per iteration, following its rules exactly.

cashu/core/base.py

@@ -19,18 +19,28 @@
 from ..mint.events.event_model import LedgerEvent
 from .crypto.aes import AESCipher
 from .crypto.b_dhke import hash_to_curve
+from .crypto.bls import PrivateKey as BlsPrivateKey
+from .crypto.bls import PublicKey as BlsPublicKey
 from .crypto.keys import (
     derive_keys,
     derive_keys_deprecated_pre_0_15,
+    derive_keys_v3,
     derive_keyset_id,
     derive_keyset_id_deprecated,
     derive_keyset_id_v2,
+    derive_keyset_id_v3,
     derive_pubkeys,
+    is_bls_keyset,
 )
-from .crypto.secp import PrivateKey, PublicKey
+from .crypto.secp import PrivateKey as SecpPrivateKey
+from .crypto.secp import PublicKey as SecpPublicKey
+from .crypto.taproot import is_taproot_point_secret
 from .legacy import derive_keys_backwards_compatible_insecure_pre_0_12
 from .settings import settings
 
+PrivateKey = Union[SecpPrivateKey, BlsPrivateKey]
+PublicKey = Union[SecpPublicKey, BlsPublicKey]
+
 
 class DLEQ(BaseModel):
     """
@@ -117,6 +127,22 @@ def from_witness(cls, witness: str):
         return cls(**json.loads(witness))
 
 
+class SpendInfo(BaseModel):
+    """Taproot spend info (spec 2.5): a key and, when conditions exist, the leaf tree.
+
+    `k` (32-byte scalar hex, bearer) and `E` (33-byte point hex, receiver-keyed)
+    are mutually exclusive. `K` (33-byte point hex) is the internal key, needed
+    when neither yields one: a script-only proof discloses its tree and `K` so
+    the tree can be checked complete. `tree` lists serialized leaves (hex) in
+    slot-map order. Local-only: never sent to the mint.
+    """
+
+    k: Optional[str] = None
+    E: Optional[str] = None
+    K: Optional[str] = None
+    tree: Optional[List[str]] = None
+
+
 class Proof(BaseModel):
     """
     Value token
@@ -129,6 +155,7 @@ class Proof(BaseModel):
     C: str = ""  # signature on secret, unblinded by wallet
     dleq: Optional[DLEQWallet] = None  # DLEQ proof
     witness: Union[None, str] = None  # witness for spending condition
+    spend_info: Optional[SpendInfo] = None  # taproot spend info (local-only)
 
     # whether this proof is reserved for sending, used for coin management in the wallet
     reserved: Union[None, bool] = False
@@ -146,7 +173,15 @@ class Proof(BaseModel):
 
     def __init__(self, **data):
         super().__init__(**data)
-        self.Y = hash_to_curve(self.secret.encode("utf-8")).format().hex()
+        if is_bls_keyset(self.id):
+            # V3: Y lives on BLS G1, hashed over the secret's raw bytes for
+            # point secrets (taproot) with utf8 fallback for legacy secrets.
+            from .crypto.bls_dhke import hash_to_curve as bls_hash_to_curve
+            from .crypto.bls_dhke import secret_to_hash_input
+
+            self.Y = bls_hash_to_curve(secret_to_hash_input(self.secret)).format().hex()
+        else:
+            self.Y = hash_to_curve(self.secret.encode("utf-8")).format().hex()
 
     @classmethod
     def from_dict(cls, proof_dict: dict):
@@ -941,12 +976,16 @@ def serialize(self):
         )
 
     @classmethod
-    def from_row(cls, row: Row):
+    def from_row(cls, row: RowMapping):
         def deserialize(serialized: str) -> Dict[int, PublicKey]:
-            return {
-                int(amount): PublicKey(bytes.fromhex(hex_key))
-                for amount, hex_key in dict(json.loads(serialized)).items()
-            }
+            is_v3 = is_bls_keyset(row["id"])
+            pub_keys: Dict[int, PublicKey] = {}
+            for amount, hex_key in dict(json.loads(serialized)).items():
+                if is_v3:
+                    pub_keys[int(amount)] = BlsPublicKey(bytes.fromhex(hex_key), group="G2")
+                else:
+                    pub_keys[int(amount)] = SecpPublicKey(bytes.fromhex(hex_key))
+            return pub_keys
 
         return cls(
             id=row["id"],
@@ -1156,7 +1195,7 @@ def generate_keys(self):
                 assert self.public_keys is not None
                 self.id = derive_keyset_id(self.public_keys)
                 logger.info(f"Generated keyset v1 ID: {self.id}")
-        else:
+        elif self.version_tuple < (0, 21):
             self.private_keys = derive_keys(
                 self.seed, self.derivation_path, self.amounts
             )
@@ -1175,6 +1214,19 @@ def generate_keys(self):
                     self.input_fee_ppk,
                 )
                 logger.info(f"Generated keyset v2 ID: {self.id}")
+        else:
+            self.private_keys = derive_keys_v3(
+                self.seed, self.derivation_path, self.amounts
+            )  # type: ignore[assignment]
+            self.public_keys = derive_pubkeys(self.private_keys, self.amounts)  # type: ignore
+
+            # KEYSETS V3: BLS12-381 cryptography
+            if id_in_db:
+                self.id = id_in_db
+            else:
+                assert self.public_keys is not None
+                self.id = derive_keyset_id_v3(self.public_keys, self.unit.name, self.input_fee_ppk)  # type: ignore[arg-type]
+                logger.info(f"Generated keyset v3 (BLS) ID: {self.id}")
 
 
 # ------- TOKEN -------
@@ -1344,6 +1396,15 @@ class TokenV4DLEQ(BaseModel):
     r: bytes
 
 
+class TokenV4SpendInfo(BaseModel):
+    """Taproot spend info in a V4 token: bearer key, DH ephemeral, internal key, leaf tree."""
+
+    k: Optional[bytes] = None
+    e: Optional[bytes] = None
+    i: Optional[bytes] = None
+    t: Optional[List[bytes]] = None
+
+
 class TokenV4Proof(BaseModel):
     """
     Value token
@@ -1354,6 +1415,7 @@ class TokenV4Proof(BaseModel):
     c: bytes  # signature
     d: Optional[TokenV4DLEQ] = None  # DLEQ proof
     w: Optional[str] = None  # witness
+    si: Optional[TokenV4SpendInfo] = None  # taproot spend info
 
     @classmethod
     def from_proof(cls, proof: Proof, include_dleq=False):
@@ -1370,7 +1432,25 @@ def from_proof(cls, proof: Proof, include_dleq=False):
                 if proof.dleq
                 else None
             ),
-            w=proof.witness,
+            # A v3 witness signs one transaction's digest, so it means nothing
+            # outside that transaction and a token carries no transaction.
+            # Emitting one would hand the next owner a witness that can never
+            # verify, in place of the signature they have to produce.
+            w=None if is_taproot_point_secret(proof.secret, proof.id) else proof.witness,
+            si=(
+                TokenV4SpendInfo(
+                    k=bytes.fromhex(proof.spend_info.k) if proof.spend_info.k else None,
+                    e=bytes.fromhex(proof.spend_info.E) if proof.spend_info.E else None,
+                    i=bytes.fromhex(proof.spend_info.K) if proof.spend_info.K else None,
+                    t=(
+                        [bytes.fromhex(leaf) for leaf in proof.spend_info.tree]
+                        if proof.spend_info.tree
+                        else None
+                    ),
+                )
+                if proof.spend_info
+                else None
+            ),
         )
 
 
@@ -1440,7 +1520,25 @@ def proofs(self) -> List[Proof]:
                     if p.d
                     else None
                 ),
-                witness=p.w,
+                # A v3 witness signs one transaction's digest, so a token
+                # cannot carry a usable one. Keeping it would leave a stranger's
+                # witness in place of the signature the new owner must produce,
+                # and their sweep would be refused for it.
+                witness=(
+                    None
+                    if is_taproot_point_secret(p.s, token.i.hex())
+                    else p.w
+                ),
+                spend_info=(
+                    SpendInfo(
+                        k=p.si.k.hex() if p.si.k else None,
+                        E=p.si.e.hex() if p.si.e else None,
+                        K=p.si.i.hex() if p.si.i else None,
+                        tree=[leaf.hex() for leaf in p.si.t] if p.si.t else None,
+                    )
+                    if p.si
+                    else None
+                ),
             )
             for token in self.t
             for p in token.p
@@ -1507,6 +1605,17 @@ def serialize_to_dict(self, include_dleq=False):
             for proof in token["p"]:
                 if not proof.get("w"):
                     del proof["w"]
+        # strip absent spend info; drop None subfields from present ones
+        for token in return_dict["t"]:
+            for proof in token["p"]:
+                if not proof.get("si"):
+                    proof.pop("si", None)
+                else:
+                    proof["si"] = {
+                        key: value
+                        for key, value in proof["si"].items()
+                        if value is not None
+                    }
         # optional memo
         if self.d:
             return_dict.update(dict(d=self.d))

cashu/core/constants.py

@@ -1,7 +1,8 @@
 # Maximum lengths for Pydantic string fields
 MAX_UNIT_LEN = 64
-MAX_PUBKEY_LEN = 66
+MAX_PUBKEY_LEN = 96
 MAX_SIG_LEN = 130
+MAX_WITNESS_LEN = 4096
 MAX_QUOTE_ID_LEN = 256
 MAX_INVOICE_DESC_LEN = 1024
 MAX_PAYMENT_REQUEST_LEN = 10000

cashu/core/crypto/b_dhke.py

@@ -165,6 +165,8 @@ def derive_dleq_nonce(
 def step2_bob_dleq(
     B_: PublicKey, a: PrivateKey, p_bytes: bytes = b""
 ) -> Tuple[PrivateKey, PrivateKey]:
+    if not isinstance(a, PrivateKey):
+        raise TypeError(f"Expected SecpPrivateKey, got {type(a)}")
     C_: PublicKey = B_ * a  # type: ignore
     A = a.public_key
     assert A
@@ -179,7 +181,10 @@ def step2_bob_dleq(
     assert R1
     R2: PublicKey = B_ * p  # type: ignore
     e = hash_e(R1, R2, A, C_)  # e = hash(R1, R2, A, C_)
-    s = p.add(bytes.fromhex(a.multiply(e).to_hex()))  # s = p + ek
+    if isinstance(a, PrivateKey):
+        s = p.add(bytes.fromhex(a.multiply(e).to_hex()))  # s = p + ek
+    else:
+        raise TypeError(f"Expected SecpPrivateKey, got {type(a)}")
     spk = PrivateKey(bytes.fromhex(s.to_hex()))
     epk = PrivateKey(e)
     return epk, spk

cashu/core/crypto/bls.py

@@ -0,0 +1,81 @@
+import os
+from typing import Optional
+
+import pyblst
+
+curve_order = 52435875175126190479447740508185965837690552500527637822603658699938581184513
+_G2_HEX = '93e02b6052719f607dacd3a088274f65596bd0d09920b61ab5da61bbdc7f5049334cf11213945d57e5ac7d055d042b7e024aa2b2f08f0a91260805272dc51051c6e47ad4fa403b02b4510b647ae3d1770bac0326a805bbefd48056c8c121bdb8'
+G2 = pyblst.BlstP2Element().uncompress(bytes.fromhex(_G2_HEX))
+
+
+class PrivateKey:
+    def __init__(self, privkey: bytes = b"", scalar: Optional[int] = None):
+        if scalar is not None:
+            if not 0 < scalar < curve_order:
+                raise ValueError("BLS scalar must be in Fr*")
+            self.scalar = scalar
+        elif privkey:
+            self.scalar = int.from_bytes(privkey, "big")
+            if len(privkey) != 32 or not 0 < self.scalar < curve_order:
+                raise ValueError("BLS private key must be 32 bytes in Fr*")
+        else:
+            self.scalar = 0
+            while not 0 < self.scalar < curve_order:
+                self.scalar = int.from_bytes(os.urandom(32), "big")
+
+    @property
+    def private_key(self) -> bytes:
+        return self.scalar.to_bytes(32, "big")
+
+    def to_hex(self) -> str:
+        return self.private_key.hex()
+
+    def get_g2_public_key(self) -> "PublicKey":
+        pt = G2.scalar_mul(self.scalar)
+        return PublicKey(point=pt, group="G2")
+
+    @property
+    def public_key(self) -> "PublicKey":
+        return self.get_g2_public_key()
+
+
+class PublicKey:
+    def __init__(self, compressed: bytes = b"", point=None, group="G1"):
+        self.group = group
+        try:
+            if point is not None:
+                self.point = point
+            elif compressed:
+                if self.group == "G1":
+                    self.point = pyblst.BlstP1Element().uncompress(compressed)
+                else:
+                    self.point = pyblst.BlstP2Element().uncompress(compressed)
+            else:
+                raise ValueError("Must provide point or compressed bytes")
+        except Exception:
+            raise ValueError("The public key could not be parsed or is invalid.")
+
+    def format(self, compressed: bool = True) -> bytes:
+        return self.point.compress()
+
+    def serialize(self) -> bytes:
+        return self.format()
+
+    def is_infinity(self) -> bool:
+        """Check if the point is the point at infinity (additive identity)."""
+        if self.group == "G1":
+            return self.point == pyblst.BlstP1Element()
+        else:
+            return self.point == pyblst.BlstP2Element()
+
+    def __eq__(self, other):
+        if isinstance(other, PublicKey):
+            return self.point == other.point
+        return False
+
+    def __mul__(self, scalar):
+        if isinstance(scalar, PrivateKey):
+            return PublicKey(point=self.point.scalar_mul(scalar.scalar), group=self.group)
+        elif isinstance(scalar, int):
+            return PublicKey(point=self.point.scalar_mul(scalar), group=self.group)
+        raise TypeError("Can't multiply with non-scalar")