wip: done taurus in mem and pubsub

Author: vietddude

pkg/eventconsumer/event_consumer.go

@@ -2,7 +2,6 @@ package eventconsumer
 
 import (
 	"context"
-	"encoding/hex"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -187,7 +186,6 @@ func (ec *eventConsumer) handleKeyGenEvent(natMsg *nats.Msg) {
 
 	// ecdsaSession.Init()
 	// eddsaSession.Init()
-	taurusSession.Init()
 
 	// ctxEcdsa, doneEcdsa := context.WithCancel(baseCtx)
 	// ctxEddsa, doneEddsa := context.WithCancel(baseCtx)
@@ -225,66 +223,32 @@ func (ec *eventConsumer) handleKeyGenEvent(natMsg *nats.Msg) {
 	// 	}
 	// }()
 
-	go func() {
-		select {
-		case <-ctxTaurus.Done():
-			return
-		case err := <-taurusSession.ErrChan():
-			if err != nil {
-				logger.Error("CGGMP21 keygen session error", err)
-				errorChan <- err
-				doneTaurus()
-			}
-		}
-	}()
-
 	// ecdsaSession.ListenToIncomingMessageAsync()
 	// eddsaSession.ListenToIncomingMessageAsync()
-	taurusSession.ListenToIncomingMessageAsync(taurusSession.ProcessInboundMessage)
 	// Temporary delay for peer setup
 	ec.warmUpSession()
 	// go ecdsaSession.GenerateKey(doneEcdsa)
 	// go eddsaSession.GenerateKey(doneEddsa)
-	go taurusSession.ProcessOutboundMessage()
 
-	// Wait for the keygen to complete
-	completionChan := make(chan string, 1)
 	go func() {
-		result := taurusSession.WaitForFinish()
-		completionChan <- result
-	}()
-
-	// Wait for completion, error, or timeout
-	select {
-	case pubKeyHex := <-completionChan:
-		// Success - set the public key
-		if pubKeyHex != "" {
-			pubKeyBytes, err := hex.DecodeString(pubKeyHex)
-			if err == nil {
-				successEvent.TaurusCMPPubKey = pubKeyBytes
-			}
+		data, err := taurusSession.Keygen(ctxTaurus)
+		if err != nil {
+			logger.Error("Failed to generate key", err)
+			ec.handleKeygenSessionError(walletID, err, "Failed to generate key", natMsg)
+			errorChan <- err
+			doneTaurus()
 		}
-		doneTaurus() // Signal completion
-
-	case err := <-errorChan:
-		// Error occurred
-		ec.handleKeygenSessionError(walletID, err, "CGGMP21 keygen error", natMsg)
-		return
-
-	case <-baseCtx.Done():
-		// Timeout occurred
-		logger.Warn("Key generation timed out", "walletID", walletID, "timeout", KeyGenTimeOut)
-		ec.handleKeygenSessionError(walletID, fmt.Errorf("keygen session timed out after %v", KeyGenTimeOut), "Key generation timed out", natMsg)
-		return
-	}
+		successEvent.TaurusCMPPubKey = data.Payload
+		doneTaurus()
+	}()
 
 	payload, err := json.Marshal(successEvent)
 	if err != nil {
 		logger.Error("Failed to marshal keygen success event", err)
 		ec.handleKeygenSessionError(walletID, err, "Failed to marshal keygen success event", natMsg)
 		return
 	}
-
+	fmt.Println("payload", string(payload))
 	key := fmt.Sprintf(mpc.TypeGenerateWalletResultFmt, walletID)
 	if err := ec.genKeyResultQueue.Enqueue(
 		key,

pkg/mpc/node.go

@@ -16,6 +16,7 @@ import (
 	"github.com/fystack/mpcium/pkg/messaging"
 	"github.com/fystack/mpcium/pkg/mpc/taurus"
 	"github.com/taurusgroup/multi-party-sig/pkg/party"
+	"github.com/taurusgroup/multi-party-sig/pkg/pool"
 )
 
 const (
@@ -146,32 +147,14 @@ func (p *Node) CreateCMPKeyGenSession(
 	walletID string,
 	threshold int,
 	resultQueue messaging.MessageQueue,
-) (taurus.KeyGenSession, error) {
-	if !p.peerRegistry.ArePeersReady() {
-		return nil, fmt.Errorf(
-			"peers are not ready yet. ready: %d, expected: %d",
-			p.peerRegistry.GetReadyPeersCount(),
-			len(p.peerIDs)+1,
-		)
-	}
-
+) (*taurus.CmpParty, error) {
 	readyPeerIDs := p.peerRegistry.GetReadyPeersIncludeSelf()
 	selfPartyID, allPartyIDs := p.generateTaurusPartyIDs(PurposeKeygen, readyPeerIDs, DefaultVersion)
-
-	session := taurus.NewCGGMP21KeygenSession(
-		walletID,
-		p.pubSub,
-		selfPartyID,
-		allPartyIDs,
-		threshold,
-		p.kvstore,
-		p.keyinfoStore,
-		resultQueue,
-		p.identityStore,
-	)
-
-	session.Init()
-	return session, nil
+	tr := taurus.NewNATSTransport(walletID, selfPartyID, p.pubSub)
+	adapter := taurus.NewTaurusNetworkAdapter(walletID, selfPartyID, tr, allPartyIDs)
+	pl := pool.NewPool(0)
+	party := taurus.NewCmpParty(walletID, selfPartyID, allPartyIDs, threshold, pl, adapter)
+	return party, nil
 }
 
 func (p *Node) CreateSigningSession(
@@ -506,8 +489,8 @@ func sessionKeyPrefix(sessionType SessionType) (string, error) {
 	}
 }
 
-func (p *Node) generateTaurusPartyIDs(purpose string, peerIDs []string, version int) (party.ID, []party.ID) {
-	partyIDs := make([]party.ID, len(peerIDs))
+func (p *Node) generateTaurusPartyIDs(purpose string, peerIDs []string, version int) (party.ID, party.IDSlice) {
+	partyIDs := make(party.IDSlice, len(peerIDs))
 	var selfPartyID party.ID
 
 	for i, peerID := range peerIDs {

pkg/mpc/taurus/adapter.go

@@ -0,0 +1,87 @@
+package taurus
+
+import (
+	"encoding/json"
+	"log/slog"
+
+	"github.com/taurusgroup/multi-party-sig/pkg/party"
+	"github.com/taurusgroup/multi-party-sig/pkg/protocol"
+)
+
+type NetworkInterface interface {
+	Next() <-chan *protocol.Message
+	Send(msg *protocol.Message)
+	Done() <-chan struct{}
+}
+
+type TaurusNetworkAdapter struct {
+	sid       string
+	selfID    party.ID
+	transport Transport
+	inbox     chan *protocol.Message
+	done      chan struct{}
+	peers     party.IDSlice
+}
+
+func NewTaurusNetworkAdapter(
+	sid string,
+	selfID party.ID,
+	t Transport,
+	peers party.IDSlice,
+) *TaurusNetworkAdapter {
+	a := &TaurusNetworkAdapter{
+		sid:       sid,
+		selfID:    selfID,
+		transport: t,
+		inbox:     make(chan *protocol.Message, 100),
+		done:      make(chan struct{}),
+		peers:     peers,
+	}
+	go a.route()
+	return a
+}
+
+func (a *TaurusNetworkAdapter) Next() <-chan *protocol.Message { return a.inbox }
+func (a *TaurusNetworkAdapter) Done() <-chan struct{}          { return a.done }
+
+func (a *TaurusNetworkAdapter) Send(msg *protocol.Message) {
+	wire, err := json.Marshal(msg)
+	if err != nil {
+		slog.Error("❌ marshal protocol msg", "err", err)
+		return
+	}
+	m := Msg{SID: a.sid, From: string(msg.From), IsBroadcast: msg.Broadcast, Data: wire}
+	for _, pid := range a.peers {
+		if pid == a.selfID {
+			continue
+		}
+		if msg.Broadcast || msg.IsFor(pid) {
+			_ = a.transport.Send(string(pid), m)
+		}
+	}
+}
+
+func (a *TaurusNetworkAdapter) route() {
+	for {
+		select {
+		case tm, ok := <-a.transport.Inbox():
+			if !ok {
+				close(a.done)
+				return
+			}
+			var pm protocol.Message
+			if err := json.Unmarshal(tm.Data, &pm); err != nil {
+				slog.Error("❌ unmarshal protocol msg", "err", err)
+				continue
+			}
+			select {
+			case a.inbox <- &pm:
+			default:
+				slog.Warn("⚠️ inbox full, drop msg", "self", a.selfID)
+			}
+		case <-a.transport.Done():
+			close(a.done)
+			return
+		}
+	}
+}

pkg/mpc/taurus/cmp.go

@@ -0,0 +1,152 @@
+package taurus
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"math/big"
+
+	"github.com/fystack/mpcium/pkg/logger"
+	"github.com/fystack/mpcium/pkg/types"
+	"github.com/taurusgroup/multi-party-sig/pkg/ecdsa"
+	"github.com/taurusgroup/multi-party-sig/pkg/math/curve"
+	"github.com/taurusgroup/multi-party-sig/pkg/party"
+	"github.com/taurusgroup/multi-party-sig/pkg/pool"
+	"github.com/taurusgroup/multi-party-sig/pkg/protocol"
+	"github.com/taurusgroup/multi-party-sig/protocols/cmp"
+)
+
+type CmpParty struct {
+	sid       string
+	id        party.ID
+	ids       party.IDSlice
+	threshold int
+	pl        *pool.Pool
+	savedData *cmp.Config
+	network   NetworkInterface
+}
+
+func NewCmpParty(
+	sid string,
+	id party.ID,
+	ids party.IDSlice,
+	threshold int,
+	pl *pool.Pool,
+	network NetworkInterface,
+) *CmpParty {
+	return &CmpParty{
+		sid:       sid,
+		id:        id,
+		ids:       ids,
+		threshold: threshold,
+		pl:        pl,
+		network:   network,
+	}
+}
+
+func (p *CmpParty) LoadKey(data *types.KeyData) error {
+	cfg := cmp.EmptyConfig(curve.Secp256k1{})
+	if err := cfg.UnmarshalBinary(data.Payload); err != nil {
+		return fmt.Errorf("decode key data: %w", err)
+	}
+	p.savedData = cfg
+	return nil
+}
+
+func (p *CmpParty) Keygen(ctx context.Context) (types.KeyData, error) {
+	h, err := protocol.NewMultiHandler(
+		cmp.Keygen(curve.Secp256k1{}, p.id, p.ids, p.threshold, p.pl),
+		[]byte(p.sid),
+	)
+	if err != nil {
+		return types.KeyData{}, err
+	}
+	if err := p.executeProtocol(ctx, h); err != nil {
+		return types.KeyData{}, err
+	}
+	res, err := h.Result()
+	if err != nil {
+		return types.KeyData{}, err
+	}
+	cfg, ok := res.(*cmp.Config)
+	if !ok {
+		return types.KeyData{}, errors.New("unexpected result type")
+	}
+	p.savedData = cfg
+	packed, _ := cfg.MarshalBinary()
+	return types.KeyData{SID: p.sid, Type: "taurus_cmp", Payload: packed}, nil
+}
+
+func (p *CmpParty) Sign(ctx context.Context, msg *big.Int) ([]byte, error) {
+	if p.savedData == nil {
+		return nil, errors.New("no key loaded")
+	}
+	h, err := protocol.NewMultiHandler(
+		cmp.Sign(p.savedData, p.ids, msg.Bytes(), p.pl),
+		[]byte(p.sid),
+	)
+	if err != nil {
+		return nil, err
+	}
+	if err := p.executeProtocol(ctx, h); err != nil {
+		return nil, err
+	}
+	res, err := h.Result()
+	if err != nil {
+		return nil, err
+	}
+	sig, ok := res.(*ecdsa.Signature)
+	if !ok {
+		return nil, errors.New("unexpected signature result")
+	}
+	if !sig.Verify(p.savedData.PublicPoint(), msg.Bytes()) {
+		return nil, errors.New("failed to verify cmp signature")
+	}
+	return sig.SigEthereum()
+}
+
+func (p *CmpParty) Reshare(ctx context.Context) (types.KeyData, error) {
+	if p.savedData == nil {
+		return types.KeyData{}, errors.New("no key loaded")
+	}
+	h, err := protocol.NewMultiHandler(cmp.Refresh(p.savedData, p.pl), []byte(p.sid))
+	if err != nil {
+		return types.KeyData{}, err
+	}
+	if err := p.executeProtocol(ctx, h); err != nil {
+		return types.KeyData{}, err
+	}
+	res, err := h.Result()
+	if err != nil {
+		return types.KeyData{}, err
+	}
+	cfg, ok := res.(*cmp.Config)
+	if !ok {
+		return types.KeyData{}, errors.New("unexpected result type")
+	}
+	p.savedData = cfg
+	packed, _ := cfg.MarshalBinary()
+	return types.KeyData{SID: p.sid, Type: "taurus_cmp", Payload: packed}, nil
+}
+
+func (p *CmpParty) executeProtocol(ctx context.Context, h protocol.Handler) error {
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case msg, ok := <-h.Listen():
+			if !ok {
+				return nil
+			}
+			p.network.Send(msg)
+		case msg := <-p.network.Next():
+			if h.CanAccept(msg) {
+				h.Accept(msg)
+			} else {
+				logger.Warn("⚠️ Ignored invalid msg", "self", p.id)
+			}
+		case <-p.network.Done():
+			return nil
+		}
+	}
+}