From 34f8ad5928a001a1a436bbeb6a97d5ac6d20cbde Mon Sep 17 00:00:00 2001 From: vietddude Date: Mon, 26 Oct 2026 10:52:20 +0700 Subject: [PATCH] Verify node-signed results with t+1 threshold Nodes sign keygen, signing and reshare result events with their identity key. The client verifies each signature and only delivers a result once t+1 distinct nodes report the same outcome. --- cmd/mpcium/main.go | 5 +- e2e/base_test.go | 34 +++- e2e/go.mod | 6 +- e2e/go.sum | 12 +- e2e/keygen_test.go | 3 +- e2e/reshare_test.go | 9 +- e2e/sign_test.go | 5 +- examples/generate/main.go | 10 +- examples/reshare/main.go | 10 +- examples/sign/main.go | 10 +- pkg/client/client.go | 114 ++++++++++- pkg/client/node_key_lookup_test.go | 50 +++++ pkg/client/result_aggregator.go | 267 ++++++++++++++++++++++++++ pkg/client/result_aggregator_test.go | 151 +++++++++++++++ pkg/event/keygen.go | 5 + pkg/event/reshare.go | 6 + pkg/event/result.go | 101 ++++++++++ pkg/event/result_test.go | 87 +++++++++ pkg/event/sign.go | 6 + pkg/event/types.go | 3 + pkg/eventconsumer/event_consumer.go | 64 ++++-- pkg/eventconsumer/keygen_consumer.go | 11 +- pkg/eventconsumer/sign_consumer.go | 15 +- pkg/eventconsumer/timeout_consumer.go | 22 ++- pkg/identity/identity.go | 19 ++ pkg/mpc/ecdsa_signing_session.go | 7 +- pkg/mpc/eddsa_signing_session.go | 7 +- 27 files changed, 982 insertions(+), 57 deletions(-) create mode 100644 pkg/client/node_key_lookup_test.go create mode 100644 pkg/client/result_aggregator.go create mode 100644 pkg/client/result_aggregator_test.go create mode 100644 pkg/event/result.go create mode 100644 pkg/event/result_test.go diff --git a/cmd/mpcium/main.go b/cmd/mpcium/main.go index a68eb0d8..1e1098b2 100644 --- a/cmd/mpcium/main.go +++ b/cmd/mpcium/main.go @@ -217,12 +217,13 @@ func runNode(ctx context.Context, c *cli.Command) error { timeoutConsumer := eventconsumer.NewTimeOutConsumer( natsConn, singingResultQueue, + identityStore, ) timeoutConsumer.Run() defer timeoutConsumer.Close() - keygenConsumer := eventconsumer.NewKeygenConsumer(natsConn, keygenBroker, pubsub, peerRegistry, genKeyResultQueue) - signingConsumer := eventconsumer.NewSigningConsumer(natsConn, signingBroker, pubsub, peerRegistry, singingResultQueue) + keygenConsumer := eventconsumer.NewKeygenConsumer(natsConn, keygenBroker, pubsub, peerRegistry, genKeyResultQueue, identityStore) + signingConsumer := eventconsumer.NewSigningConsumer(natsConn, signingBroker, pubsub, peerRegistry, singingResultQueue, identityStore) // Make the node ready before starting the signing consumer if err := peerRegistry.Ready(); err != nil { diff --git a/e2e/base_test.go b/e2e/base_test.go index 5c209c4f..17f0985d 100644 --- a/e2e/base_test.go +++ b/e2e/base_test.go @@ -9,6 +9,7 @@ import ( "path/filepath" "strconv" "strings" + "sync" "syscall" "testing" "time" @@ -59,6 +60,9 @@ type E2ETestSuite struct { signingResults map[string]*event.SigningResultEvent resharingResults map[string]*event.ResharingResultEvent config TestConfig + + resultErrorsMu sync.Mutex + resultErrors []string } func NewE2ETestSuite(testDir string) *E2ETestSuite { @@ -193,9 +197,18 @@ func (s *E2ETestSuite) SetupMPCClient(t *testing.T) { t.Fatalf("Failed to create local signer: %v", err) } + // Verify node-signed result events using the identity files generated by + // setup_test_identities.sh (distributed to every test node's identity dir). + nodeKeyLookup, err := client.NewNodeKeyLookup(filepath.Join(s.testDir, "test_node0", "identity")) + if err != nil { + t.Fatalf("Failed to create node key lookup: %v", err) + } + mpcClient := client.NewMPCClient(client.Options{ - NatsConn: s.natsConn, - Signer: localSigner, + NatsConn: s.natsConn, + Signer: localSigner, + NodeKeyLookup: nodeKeyLookup, + RequiredNodeSignatures: s.config.MPCThreshold + 1, }) s.mpcClient = mpcClient t.Log("MPC client created") @@ -670,3 +683,20 @@ func (s *E2ETestSuite) CleanupTestEnvironment(t *testing.T) { t.Log("Test environment cleanup completed") } + +// recordResultError is safe to call from result callbacks that may fire after the subtest that registered them has ended. +func (s *E2ETestSuite) recordResultError(format string, args ...any) { + s.resultErrorsMu.Lock() + defer s.resultErrorsMu.Unlock() + s.resultErrors = append(s.resultErrors, fmt.Sprintf(format, args...)) +} + +func (s *E2ETestSuite) AssertNoResultErrors(t *testing.T) { + // Wait out the client's result aggregation window so late insufficient-signature errors are caught. + time.Sleep(16 * time.Second) + s.resultErrorsMu.Lock() + defer s.resultErrorsMu.Unlock() + for _, msg := range s.resultErrors { + t.Errorf("error result received: %s", msg) + } +} diff --git a/e2e/go.mod b/e2e/go.mod index 4112bb34..b41324d9 100644 --- a/e2e/go.mod +++ b/e2e/go.mod @@ -1,12 +1,12 @@ module github.com/fystack/mpcium/e2e -go 1.23.0 +go 1.23.8 require ( github.com/dgraph-io/badger/v4 v4.7.0 github.com/fystack/mpcium v0.0.0-00010101000000-000000000000 github.com/google/uuid v1.6.0 - github.com/hashicorp/consul/api v1.26.1 + github.com/hashicorp/consul/api v1.32.1 github.com/nats-io/nats.go v1.31.0 github.com/stretchr/testify v1.10.0 gopkg.in/yaml.v2 v2.4.0 @@ -91,7 +91,7 @@ require ( go.uber.org/multierr v1.9.0 // indirect go.uber.org/zap v1.21.0 // indirect golang.org/x/crypto v0.37.0 // indirect - golang.org/x/exp v0.0.0-20240112132812-db7319d0e0e3 // indirect + golang.org/x/exp v0.0.0-20250305212735-054e65f0b394 // indirect golang.org/x/net v0.39.0 // indirect golang.org/x/sys v0.33.0 // indirect golang.org/x/term v0.31.0 // indirect diff --git a/e2e/go.sum b/e2e/go.sum index 804a2c4e..62e53389 100644 --- a/e2e/go.sum +++ b/e2e/go.sum @@ -158,10 +158,10 @@ github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/ github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= -github.com/hashicorp/consul/api v1.26.1 h1:5oSXOO5fboPZeW5SN+TdGFP/BILDgBm19OrPZ/pICIM= -github.com/hashicorp/consul/api v1.26.1/go.mod h1:B4sQTeaSO16NtynqrAdwOlahJ7IUDZM9cj2420xYL8A= -github.com/hashicorp/consul/sdk v0.15.0 h1:2qK9nDrr4tiJKRoxPGhm6B7xJjLVIQqkjiab2M4aKjU= -github.com/hashicorp/consul/sdk v0.15.0/go.mod h1:r/OmRRPbHOe0yxNahLw7G9x5WG17E1BIECMtCjcPSNo= +github.com/hashicorp/consul/api v1.32.1 h1:0+osr/3t/aZNAdJX558crU3PEjVrG4x6715aZHRgceE= +github.com/hashicorp/consul/api v1.32.1/go.mod h1:mXUWLnxftwTmDv4W3lzxYCPD199iNLLUyLfLGFJbtl4= +github.com/hashicorp/consul/sdk v0.16.1 h1:V8TxTnImoPD5cj0U9Spl0TUxcytjcbbJeADFF07KdHg= +github.com/hashicorp/consul/sdk v0.16.1/go.mod h1:fSXvwxB2hmh1FMZCNl6PwX0Q/1wdWtHJcZ7Ea5tns0s= github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4= github.com/hashicorp/errwrap v1.1.0 h1:OxrOeh75EUXMY8TBjag2fzXGZ40LB6IKw45YeGUDY2I= github.com/hashicorp/errwrap v1.1.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4= @@ -402,8 +402,8 @@ golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5y golang.org/x/crypto v0.13.0/go.mod h1:y6Z2r+Rw4iayiXXAIxJIDAJ1zMW4yaTpebo8fPOliYc= golang.org/x/crypto v0.37.0 h1:kJNSjF/Xp7kU0iB2Z+9viTPMW4EqqsrywMXLJOOsXSE= golang.org/x/crypto v0.37.0/go.mod h1:vg+k43peMZ0pUMhYmVAWysMK35e6ioLh3wB8ZCAfbVc= -golang.org/x/exp v0.0.0-20240112132812-db7319d0e0e3 h1:hNQpMuAJe5CtcUqCXaWga3FHu+kQvCqcsoVaQgSV60o= -golang.org/x/exp v0.0.0-20240112132812-db7319d0e0e3/go.mod h1:idGWGoKP1toJGkd5/ig9ZLuPcZBC3ewk7SzmH0uou08= +golang.org/x/exp v0.0.0-20250305212735-054e65f0b394 h1:nDVHiLt8aIbd/VzvPWN6kSOPE7+F/fNFDSXLVYkE/Iw= +golang.org/x/exp v0.0.0-20250305212735-054e65f0b394/go.mod h1:sIifuuw/Yco/y6yb6+bDNfyeQ/MdPUy/hKEMYQV17cM= golang.org/x/lint v0.0.0-20190930215403-16217165b5de/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc= golang.org/x/mod v0.0.0-20190513183733-4bf6d317e70e/go.mod h1:mXi4GBBbnImb6dmsKGUJ2LatrhH/nqhxcFungHvyanc= golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= diff --git a/e2e/keygen_test.go b/e2e/keygen_test.go index 42085cea..83558b8e 100644 --- a/e2e/keygen_test.go +++ b/e2e/keygen_test.go @@ -21,6 +21,7 @@ func TestKeyGeneration(t *testing.T) { // Ensure cleanup happens even if test fails defer func() { + suite.AssertNoResultErrors(t) t.Log("Performing post-test cleanup...") suite.Cleanup(t) }() @@ -92,7 +93,7 @@ func testKeyGeneration(t *testing.T, suite *E2ETestSuite) { suite.keygenResults[result.WalletID] = &result if result.ResultType == event.ResultTypeError { - t.Logf("Keygen failed for wallet %s: %s (%s)", result.WalletID, result.ErrorReason, result.ErrorCode) + suite.recordResultError("Keygen failed for wallet %s: %s (%s)", result.WalletID, result.ErrorReason, result.ErrorCode) } else { t.Logf("Keygen succeeded for wallet %s", result.WalletID) } diff --git a/e2e/reshare_test.go b/e2e/reshare_test.go index f865e5d6..dbc21b4e 100644 --- a/e2e/reshare_test.go +++ b/e2e/reshare_test.go @@ -27,6 +27,7 @@ func TestResharing(t *testing.T) { // Ensure cleanup happens even if test fails defer func() { + suite.AssertNoResultErrors(t) t.Log("Performing post-test cleanup...") suite.Cleanup(t) }() @@ -36,10 +37,10 @@ func TestResharing(t *testing.T) { suite.RunMakeClean() suite.SetupInfrastructure(t) suite.SetupTestNodes(t) + require.NoError(t, suite.LoadConfig(), "Failed to load config after setup") suite.RegisterPeers(t) suite.StartNodes(t) suite.SetupMPCClient(t) - suite.LoadConfig() }) // Key generation phase @@ -84,7 +85,7 @@ func testKeyGenerationForResharing(t *testing.T, suite *E2ETestSuite) { suite.keygenResults[result.WalletID] = &result if result.ResultType == event.ResultTypeError { - t.Logf("Keygen failed for wallet %s: %s (%s)", result.WalletID, result.ErrorReason, result.ErrorCode) + suite.recordResultError("Keygen failed for wallet %s: %s (%s)", result.WalletID, result.ErrorReason, result.ErrorCode) } else { t.Logf("Keygen succeeded for wallet %s", result.WalletID) } @@ -171,7 +172,7 @@ func testResharingAllNodes(t *testing.T, suite *E2ETestSuite) { suite.resharingResults[result.WalletID] = &result if result.ResultType == event.ResultTypeError { - t.Logf("Resharing failed for wallet %s: %s (%s)", result.WalletID, result.ErrorReason, result.ErrorCode) + suite.recordResultError("Resharing failed for wallet %s: %s (%s)", result.WalletID, result.ErrorReason, result.ErrorCode) } else { t.Logf("Resharing succeeded for wallet %s", result.WalletID) } @@ -352,7 +353,7 @@ func testSigningAfterResharing(t *testing.T, suite *E2ETestSuite) { signingResults[result.TxID] = &result if result.ResultType == event.ResultTypeError { - t.Logf("Signing failed for wallet %s (tx: %s): %s (%s)", result.WalletID, result.TxID, result.ErrorReason, result.ErrorCode) + suite.recordResultError("Signing failed for wallet %s (tx: %s): %s (%s)", result.WalletID, result.TxID, result.ErrorReason, result.ErrorCode) } else { t.Logf("Signing succeeded for wallet %s (tx: %s)", result.WalletID, result.TxID) } diff --git a/e2e/sign_test.go b/e2e/sign_test.go index fe197c9e..ac681fa5 100644 --- a/e2e/sign_test.go +++ b/e2e/sign_test.go @@ -63,6 +63,7 @@ func TestSigning(t *testing.T) { // Ensure cleanup happens even if test fails defer func() { + suite.AssertNoResultErrors(t) t.Log("Performing post-test cleanup...") suite.Cleanup(t) }() @@ -142,7 +143,7 @@ func testKeyGenerationForSigning(t *testing.T, suite *E2ETestSuite) { suite.keygenResults[result.WalletID] = &result if result.ResultType == event.ResultTypeError { - t.Logf("Keygen failed for wallet %s: %s (%s)", result.WalletID, result.ErrorReason, result.ErrorCode) + suite.recordResultError("Keygen failed for wallet %s: %s (%s)", result.WalletID, result.ErrorReason, result.ErrorCode) } else { t.Logf("Keygen succeeded for wallet %s", result.WalletID) } @@ -224,7 +225,7 @@ func testSigningAllNodes(t *testing.T, suite *E2ETestSuite) { signingResults[result.TxID] = &result if result.ResultType == event.ResultTypeError { - t.Logf("Signing failed for wallet %s (tx: %s): %s (%s)", result.WalletID, result.TxID, result.ErrorReason, result.ErrorCode) + suite.recordResultError("Signing failed for wallet %s (tx: %s): %s (%s)", result.WalletID, result.TxID, result.ErrorReason, result.ErrorCode) } else { t.Logf("Signing succeeded for wallet %s (tx: %s)", result.WalletID, result.TxID) } diff --git a/examples/generate/main.go b/examples/generate/main.go index 3f0135f1..03be66ea 100644 --- a/examples/generate/main.go +++ b/examples/generate/main.go @@ -69,9 +69,15 @@ func main() { logger.Fatal("Failed to create local signer", err) } + nodeKeyLookup, err := client.NewNodeKeyLookup("./node0/identity") + if err != nil { + logger.Fatal("Failed to create node key lookup", err) + } + mpcClient := client.NewMPCClient(client.Options{ - NatsConn: natsConn, - Signer: localSigner, + NatsConn: natsConn, + Signer: localSigner, + NodeKeyLookup: nodeKeyLookup, }) var walletStartTimes sync.Map diff --git a/examples/reshare/main.go b/examples/reshare/main.go index 47c4d858..31d7f93a 100644 --- a/examples/reshare/main.go +++ b/examples/reshare/main.go @@ -60,9 +60,15 @@ func main() { logger.Fatal("Failed to create local signer", err) } + nodeKeyLookup, err := client.NewNodeKeyLookup("./node0/identity") + if err != nil { + logger.Fatal("Failed to create node key lookup", err) + } + mpcClient := client.NewMPCClient(client.Options{ - NatsConn: natsConn, - Signer: localSigner, + NatsConn: natsConn, + Signer: localSigner, + NodeKeyLookup: nodeKeyLookup, }) // 3) Listen for signing results diff --git a/examples/sign/main.go b/examples/sign/main.go index 3424610f..76e11caf 100644 --- a/examples/sign/main.go +++ b/examples/sign/main.go @@ -60,9 +60,15 @@ func main() { logger.Fatal("Failed to create local signer", err) } + nodeKeyLookup, err := client.NewNodeKeyLookup("./node0/identity") + if err != nil { + logger.Fatal("Failed to create node key lookup", err) + } + mpcClient := client.NewMPCClient(client.Options{ - NatsConn: natsConn, - Signer: localSigner, + NatsConn: natsConn, + Signer: localSigner, + NodeKeyLookup: nodeKeyLookup, }) // 2) Once wallet exists, immediately fire a SignTransaction diff --git a/pkg/client/client.go b/pkg/client/client.go index 3121bdbd..cf3b6b56 100644 --- a/pkg/client/client.go +++ b/pkg/client/client.go @@ -2,8 +2,14 @@ package client import ( "context" + "crypto/ed25519" + "encoding/hex" "encoding/json" "fmt" + "os" + "path/filepath" + "strings" + "sync" "github.com/fystack/mpcium/pkg/event" "github.com/fystack/mpcium/pkg/eventconsumer" @@ -37,6 +43,10 @@ type mpcClient struct { signResultQueue messaging.MessageQueue reshareSuccessQueue messaging.MessageQueue signer Signer + nodeKeyLookup event.NodeKeyLookup + requiredNodeSigs int + // reshareRequiredSigs maps a resharing session ID to the new committee threshold (t+1) it requested. + reshareRequiredSigs sync.Map } // Options defines configuration options for creating a new MPCClient @@ -46,6 +56,14 @@ type Options struct { // Signer for signing messages Signer Signer + + // NodeKeyLookup, when set, is used to verify that result events (keygen, + // signing, resharing) were signed by a known MPC node. Results without a + // valid node signature are rejected. See NewNodeKeyLookup. + NodeKeyLookup event.NodeKeyLookup + + // RequiredNodeSignatures is how many distinct nodes must sign the same result (t+1); <= 1 delivers the first one. + RequiredNodeSignatures int } // NewMPCClient creates a new MPC client using the provided options. @@ -54,6 +72,9 @@ func NewMPCClient(opts Options) MPCClient { if opts.Signer == nil { logger.Fatal("Signer is required", nil) } + if opts.RequiredNodeSignatures > 1 && opts.NodeKeyLookup == nil { + logger.Fatal("NodeKeyLookup is required when RequiredNodeSignatures > 1", nil) + } // 2) Create the PubSub for both publish & subscribe signingBroker, err := messaging.NewJetStreamBroker( @@ -99,6 +120,8 @@ func NewMPCClient(opts Options) MPCClient { signResultQueue: signResultQueue, reshareSuccessQueue: reshareSuccessQueue, signer: opts.Signer, + nodeKeyLookup: opts.NodeKeyLookup, + requiredNodeSigs: opts.RequiredNodeSignatures, } } @@ -132,13 +155,17 @@ func (c *mpcClient) CreateWallet(walletID string) error { // The callback will be invoked whenever a wallet creation result is received. func (c *mpcClient) OnWalletCreationResult(callback func(event event.KeygenResultEvent)) error { + aggregator := walletResultAggregator(c.requiredNodeSigs, callback) err := c.genKeySuccessQueue.Dequeue(GenerateWalletSuccessTopic, func(msg []byte) error { var event event.KeygenResultEvent err := json.Unmarshal(msg, &event) if err != nil { return err } - callback(event) + if err := c.verifyResult(event.NodeID, event.ResultSignature, &event); err != nil { + return err + } + aggregator.Add(event) return nil }) @@ -174,13 +201,17 @@ func (c *mpcClient) SignTransaction(msg *types.SignTxMessage) error { } func (c *mpcClient) OnSignResult(callback func(event event.SigningResultEvent)) error { + aggregator := signingResultAggregator(c.requiredNodeSigs, callback) err := c.signResultQueue.Dequeue(event.SigningResultCompleteTopic, func(msg []byte) error { var event event.SigningResultEvent err := json.Unmarshal(msg, &event) if err != nil { return err } - callback(event) + if err := c.verifyResult(event.NodeID, event.ResultSignature, &event); err != nil { + return err + } + aggregator.Add(event) return nil }) @@ -208,6 +239,8 @@ func (c *mpcClient) Resharing(msg *types.ResharingMessage) error { return fmt.Errorf("Resharing: marshal error: %w", err) } + c.reshareRequiredSigs.Store(msg.SessionID, msg.NewThreshold+1) + if err := c.pubsub.Publish(eventconsumer.MPCReshareEvent, bytes); err != nil { return fmt.Errorf("Resharing: publish error: %w", err) } @@ -216,6 +249,10 @@ func (c *mpcClient) Resharing(msg *types.ResharingMessage) error { func (c *mpcClient) OnResharingResult(callback func(event event.ResharingResultEvent)) error { + aggregator := reshareResultAggregator(c.reshareRequiredFor, func(e event.ResharingResultEvent) { + c.reshareRequiredSigs.Delete(e.SessionID) + callback(e) + }) err := c.reshareSuccessQueue.Dequeue(ResharingSuccessTopic, func(msg []byte) error { logger.Info("Received reshare success message", "raw", string(msg)) var event event.ResharingResultEvent @@ -224,8 +261,11 @@ func (c *mpcClient) OnResharingResult(callback func(event event.ResharingResultE logger.Error("Failed to unmarshal reshare success event", err, "raw", string(msg)) return err } + if err := c.verifyResult(event.NodeID, event.ResultSignature, &event); err != nil { + return err + } logger.Info("Deserialized reshare success event", "event", event) - callback(event) + aggregator.Add(event) return nil }) @@ -235,3 +275,71 @@ func (c *mpcClient) OnResharingResult(callback func(event event.ResharingResultE return nil } + +// verifyResult checks a node-signed result event. Failed verification returns +// messaging.ErrPermament so the message is terminated instead of redelivered. +func (c *mpcClient) verifyResult(nodeID string, signature []byte, res interface{ Raw() ([]byte, error) }) error { + if c.nodeKeyLookup == nil { + return nil + } + if err := event.VerifyNodeResult(nodeID, signature, res, c.nodeKeyLookup); err != nil { + logger.Error("Rejecting result event", err) + return messaging.ErrPermament + } + return nil +} + +// NewNodeKeyLookup builds a NodeKeyLookup from a directory of node identity +// files (_identity.json) as produced by mpcium-cli generate-identity. +func NewNodeKeyLookup(identityDir string) (event.NodeKeyLookup, error) { + entries, err := os.ReadDir(identityDir) + if err != nil { + return nil, fmt.Errorf("failed to read identity directory: %w", err) + } + + type nodeIdentity struct { + NodeID string `json:"node_id"` + PublicKey string `json:"public_key"` + } + + keys := make(map[string][]byte) + for _, entry := range entries { + if entry.IsDir() || !strings.HasSuffix(entry.Name(), "_identity.json") { + continue + } + data, err := os.ReadFile(filepath.Join(identityDir, entry.Name())) + if err != nil { + return nil, fmt.Errorf("failed to read %s: %w", entry.Name(), err) + } + var identity nodeIdentity + if err := json.Unmarshal(data, &identity); err != nil { + return nil, fmt.Errorf("failed to parse %s: %w", entry.Name(), err) + } + publicKey, err := hex.DecodeString(identity.PublicKey) + if err != nil { + return nil, fmt.Errorf("invalid public key in %s: %w", entry.Name(), err) + } + if len(publicKey) != ed25519.PublicKeySize { + return nil, fmt.Errorf("invalid public key size in %s: %d", entry.Name(), len(publicKey)) + } + keys[identity.NodeID] = publicKey + } + + return func(nodeID string) ([]byte, error) { + publicKey, ok := keys[nodeID] + if !ok { + return nil, fmt.Errorf("unknown node: %s", nodeID) + } + return publicKey, nil + }, nil +} + +func (c *mpcClient) reshareRequiredFor(e event.ResharingResultEvent) int { + if c.requiredNodeSigs <= 1 { + return c.requiredNodeSigs + } + if required, ok := c.reshareRequiredSigs.Load(e.SessionID); ok { + return required.(int) + } + return c.requiredNodeSigs +} diff --git a/pkg/client/node_key_lookup_test.go b/pkg/client/node_key_lookup_test.go new file mode 100644 index 00000000..c2533a49 --- /dev/null +++ b/pkg/client/node_key_lookup_test.go @@ -0,0 +1,50 @@ +package client + +import ( + "crypto/ed25519" + "crypto/rand" + "encoding/hex" + "encoding/json" + "os" + "path/filepath" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestNewNodeKeyLookup(t *testing.T) { + dir := t.TempDir() + + writeIdentity := func(name, nodeID string, pub ed25519.PublicKey) { + data, err := json.Marshal(map[string]string{ + "node_name": name, + "node_id": nodeID, + "public_key": hex.EncodeToString(pub), + }) + require.NoError(t, err) + require.NoError(t, os.WriteFile(filepath.Join(dir, name+"_identity.json"), data, 0600)) + } + + pub0, _, err := ed25519.GenerateKey(rand.Reader) + require.NoError(t, err) + pub1, _, err := ed25519.GenerateKey(rand.Reader) + require.NoError(t, err) + writeIdentity("node0", "id-0", pub0) + writeIdentity("node1", "id-1", pub1) + // Non-identity files in the same directory must be ignored. + require.NoError(t, os.WriteFile(filepath.Join(dir, "node0_private.key"), []byte("x"), 0600)) + + lookup, err := NewNodeKeyLookup(dir) + require.NoError(t, err) + + got, err := lookup("id-0") + require.NoError(t, err) + require.Equal(t, []byte(pub0), got) + + got, err = lookup("id-1") + require.NoError(t, err) + require.Equal(t, []byte(pub1), got) + + _, err = lookup("unknown") + require.Error(t, err) +} diff --git a/pkg/client/result_aggregator.go b/pkg/client/result_aggregator.go new file mode 100644 index 00000000..34f28d4f --- /dev/null +++ b/pkg/client/result_aggregator.go @@ -0,0 +1,267 @@ +package client + +import ( + "bytes" + "fmt" + "sync" + "time" + + "github.com/fystack/mpcium/pkg/event" + "github.com/fystack/mpcium/pkg/logger" +) + +const defaultResultAggregateTimeout = 15 * time.Second + +// resultAggregatorConfig wires a result event type into the aggregator. +type resultAggregatorConfig[T any] struct { + // required is the number of distinct nodes that must report the same + // outcome before it is delivered (t+1). + required int + // requiredOf, when set, overrides required per result. + requiredOf func(T) int + // timeout is how long to wait for the remaining nodes after the first + // verified result arrives. + timeout time.Duration + + keyOf func(T) string + nodeOf func(T) (string, []byte) + rawOf func(T) ([]byte, error) + isSuccess func(T) bool + // deliver hands a final result to the user callback. + deliver func(T) + // insufficient builds the error event delivered when fewer than + // required nodes signed the same success result before the timeout. + insufficient func(first T, got, required int) T +} + +type pendingResult[T any] struct { + required int + + success T + hasSuccess bool + successRaw []byte + successNodes map[string]struct{} + + firstError T + hasError bool + errorNodes map[string]struct{} + + timer *time.Timer +} + +// resultAggregator buffers node-signed result events and delivers the user +// callback exactly once per operation: when `required` distinct known nodes +// report the same success, when `required` nodes report an error, or when the +// timeout fires. A single compromised node cannot force either outcome. +// Events must already be signature-verified by the caller. +type resultAggregator[T any] struct { + cfg resultAggregatorConfig[T] + mu sync.Mutex + pending map[string]*pendingResult[T] + // completed holds recently finalized keys so late results from the remaining nodes are dropped. + completed map[string]struct{} +} + +func newResultAggregator[T any](cfg resultAggregatorConfig[T]) *resultAggregator[T] { + if cfg.timeout <= 0 { + cfg.timeout = defaultResultAggregateTimeout + } + return &resultAggregator[T]{cfg: cfg, pending: make(map[string]*pendingResult[T]), completed: make(map[string]struct{})} +} + +func (a *resultAggregator[T]) requiredFor(ev T) int { + if a.cfg.requiredOf != nil { + return a.cfg.requiredOf(ev) + } + return a.cfg.required +} + +// Add feeds one verified result event into the aggregator. It may call the +// deliver callback (from the calling goroutine, or from a timer goroutine +// when the aggregate times out). +func (a *resultAggregator[T]) Add(ev T) { + cfg := a.cfg + key := cfg.keyOf(ev) + nodeID, _ := cfg.nodeOf(ev) + + a.mu.Lock() + if _, done := a.completed[key]; done { + a.mu.Unlock() + return + } + p, ok := a.pending[key] + if !ok { + p = &pendingResult[T]{ + required: a.requiredFor(ev), + successNodes: make(map[string]struct{}), + errorNodes: make(map[string]struct{}), + } + if p.required > 1 { + p.timer = time.AfterFunc(cfg.timeout, func() { a.expire(key) }) + } + a.pending[key] = p + } + + if !cfg.isSuccess(ev) { + // Error payloads legitimately differ per node, so only the count of + // distinct reporting nodes matters; the first error is delivered. + if _, dup := p.errorNodes[nodeID]; dup { + a.mu.Unlock() + return + } + p.errorNodes[nodeID] = struct{}{} + if !p.hasError { + p.firstError, p.hasError = ev, true + } + if len(p.errorNodes) < p.required { + a.mu.Unlock() + return + } + a.finish(key, p) + a.mu.Unlock() + cfg.deliver(p.firstError) + return + } + + raw, err := cfg.rawOf(ev) + if err != nil { + a.mu.Unlock() + logger.Error("Failed to canonicalize result event for aggregation", err) + return + } + if !p.hasSuccess { + p.success, p.successRaw, p.hasSuccess = ev, raw, true + } else if !bytes.Equal(raw, p.successRaw) { + a.mu.Unlock() + logger.Warn("Conflicting result events for the same operation; dropping node's event", + "key", key, "nodeID", nodeID) + return + } + if _, dup := p.successNodes[nodeID]; dup { + a.mu.Unlock() + return + } + p.successNodes[nodeID] = struct{}{} + if len(p.successNodes) < p.required { + a.mu.Unlock() + return + } + a.finish(key, p) + a.mu.Unlock() + cfg.deliver(p.success) +} + +// finish must be called with a.mu held. +func (a *resultAggregator[T]) finish(key string, p *pendingResult[T]) { + if p.timer != nil { + p.timer.Stop() + } + delete(a.pending, key) + a.markCompleted(key) +} + +// expire delivers an error result when not enough nodes agreed in time. +func (a *resultAggregator[T]) expire(key string) { + a.mu.Lock() + p, ok := a.pending[key] + if !ok { + a.mu.Unlock() + return + } + a.finish(key, p) + a.mu.Unlock() + if p.hasSuccess { + a.cfg.deliver(a.cfg.insufficient(p.success, len(p.successNodes), p.required)) + return + } + a.cfg.deliver(p.firstError) +} + +// signingResultAggregator assembles the aggregator for signing results. +func signingResultAggregator(required int, deliver func(event.SigningResultEvent)) *resultAggregator[event.SigningResultEvent] { + return newResultAggregator(resultAggregatorConfig[event.SigningResultEvent]{ + required: required, + keyOf: func(e event.SigningResultEvent) string { return e.TxID }, + nodeOf: func(e event.SigningResultEvent) (string, []byte) { + return e.NodeID, e.ResultSignature + }, + rawOf: func(e event.SigningResultEvent) ([]byte, error) { return e.Raw() }, + isSuccess: func(e event.SigningResultEvent) bool { + return e.ResultType == event.ResultTypeSuccess + }, + deliver: deliver, + insufficient: func(e event.SigningResultEvent, got, required int) event.SigningResultEvent { + e.ResultType = event.ResultTypeError + e.ErrorCode = event.ErrorCodeInsufficientNodeSignatures + e.R, e.S, e.SignatureRecovery, e.Signature = nil, nil, nil, nil + e.ErrorReason = fmt.Sprintf( + "result was signed by %d node(s) but %d are required; possible node failure or compromise", + got, required, + ) + return e + }, + }) +} + +// walletResultAggregator assembles the aggregator for keygen results. +func walletResultAggregator(required int, deliver func(event.KeygenResultEvent)) *resultAggregator[event.KeygenResultEvent] { + return newResultAggregator(resultAggregatorConfig[event.KeygenResultEvent]{ + required: required, + keyOf: func(e event.KeygenResultEvent) string { return e.WalletID }, + nodeOf: func(e event.KeygenResultEvent) (string, []byte) { + return e.NodeID, e.ResultSignature + }, + rawOf: func(e event.KeygenResultEvent) ([]byte, error) { return e.Raw() }, + isSuccess: func(e event.KeygenResultEvent) bool { + return e.ResultType == event.ResultTypeSuccess + }, + deliver: deliver, + insufficient: func(e event.KeygenResultEvent, got, required int) event.KeygenResultEvent { + e.ResultType = event.ResultTypeError + e.ECDSAPubKey, e.EDDSAPubKey = nil, nil + e.ErrorCode = string(event.ErrorCodeInsufficientNodeSignatures) + e.ErrorReason = fmt.Sprintf( + "result was signed by %d node(s) but %d are required; possible node failure or compromise", + got, required, + ) + return e + }, + }) +} + +// reshareResultAggregator assembles the aggregator for resharing results. +// Only new-committee nodes report success, so requiredOf should reflect the requested new threshold. +func reshareResultAggregator(requiredOf func(event.ResharingResultEvent) int, deliver func(event.ResharingResultEvent)) *resultAggregator[event.ResharingResultEvent] { + return newResultAggregator(resultAggregatorConfig[event.ResharingResultEvent]{ + requiredOf: requiredOf, + keyOf: func(e event.ResharingResultEvent) string { return e.WalletID + ":" + e.SessionID }, + nodeOf: func(e event.ResharingResultEvent) (string, []byte) { + return e.NodeID, e.ResultSignature + }, + rawOf: func(e event.ResharingResultEvent) ([]byte, error) { return e.Raw() }, + isSuccess: func(e event.ResharingResultEvent) bool { + return e.ResultType == event.ResultTypeSuccess + }, + deliver: deliver, + insufficient: func(e event.ResharingResultEvent, got, required int) event.ResharingResultEvent { + e.ResultType = event.ResultTypeError + e.PubKey = nil + e.ErrorCode = string(event.ErrorCodeInsufficientNodeSignatures) + e.ErrorReason = fmt.Sprintf( + "result was signed by %d node(s) but %d are required; possible node failure or compromise", + got, required, + ) + return e + }, + }) +} + +// markCompleted must be called with a.mu held. +func (a *resultAggregator[T]) markCompleted(key string) { + a.completed[key] = struct{}{} + time.AfterFunc(a.cfg.timeout, func() { + a.mu.Lock() + delete(a.completed, key) + a.mu.Unlock() + }) +} diff --git a/pkg/client/result_aggregator_test.go b/pkg/client/result_aggregator_test.go new file mode 100644 index 00000000..a6822265 --- /dev/null +++ b/pkg/client/result_aggregator_test.go @@ -0,0 +1,151 @@ +package client + +import ( + "sync" + "testing" + "time" + + "github.com/fystack/mpcium/pkg/event" +) + +func newTestSigningAggregator(required int, timeout time.Duration) (*resultAggregator[event.SigningResultEvent], func() []event.SigningResultEvent) { + var mu sync.Mutex + var delivered []event.SigningResultEvent + agg := signingResultAggregator(required, func(e event.SigningResultEvent) { + mu.Lock() + delivered = append(delivered, e) + mu.Unlock() + }) + agg.cfg.timeout = timeout + return agg, func() []event.SigningResultEvent { + mu.Lock() + defer mu.Unlock() + return append([]event.SigningResultEvent(nil), delivered...) + } +} + +func successResult(nodeID string) event.SigningResultEvent { + return event.SigningResultEvent{ + ResultType: event.ResultTypeSuccess, + TxID: "tx1", + R: []byte{1}, + S: []byte{2}, + NodeID: nodeID, + } +} + +func TestResultAggregatorDeliversOnceAndDropsLateResults(t *testing.T) { + agg, delivered := newTestSigningAggregator(2, 50*time.Millisecond) + + agg.Add(successResult("node0")) + if got := len(delivered()); got != 0 { + t.Fatalf("delivered %d results before threshold", got) + } + agg.Add(successResult("node1")) + agg.Add(successResult("node2")) + time.Sleep(100 * time.Millisecond) + + got := delivered() + if len(got) != 1 || got[0].ResultType != event.ResultTypeSuccess { + t.Fatalf("expected exactly one success delivery, got %+v", got) + } +} + +func TestResultAggregatorIgnoresDuplicateNode(t *testing.T) { + agg, delivered := newTestSigningAggregator(2, 50*time.Millisecond) + + agg.Add(successResult("node0")) + agg.Add(successResult("node0")) + time.Sleep(100 * time.Millisecond) + + got := delivered() + if len(got) != 1 || got[0].ErrorCode != event.ErrorCodeInsufficientNodeSignatures { + t.Fatalf("expected one insufficient-signatures error, got %+v", got) + } +} + +func TestResultAggregatorDropsConflictingResult(t *testing.T) { + agg, delivered := newTestSigningAggregator(2, 50*time.Millisecond) + + agg.Add(successResult("node0")) + forged := successResult("node1") + forged.R = []byte{9} + agg.Add(forged) + time.Sleep(100 * time.Millisecond) + + got := delivered() + if len(got) != 1 || got[0].ErrorCode != event.ErrorCodeInsufficientNodeSignatures { + t.Fatalf("expected one insufficient-signatures error, got %+v", got) + } +} + +func errorResult(nodeID, reason string) event.SigningResultEvent { + return event.SigningResultEvent{ + ResultType: event.ResultTypeError, + TxID: "tx1", + ErrorReason: reason, + NodeID: nodeID, + } +} + +func TestResultAggregatorSingleErrorDoesNotPreemptSuccess(t *testing.T) { + agg, delivered := newTestSigningAggregator(2, 50*time.Millisecond) + + agg.Add(errorResult("node0", "max deliveries")) + agg.Add(successResult("node1")) + agg.Add(successResult("node2")) + time.Sleep(100 * time.Millisecond) + + got := delivered() + if len(got) != 1 || got[0].ResultType != event.ResultTypeSuccess { + t.Fatalf("expected exactly one success delivery, got %+v", got) + } +} + +func TestResultAggregatorDeliversErrorReportedByRequiredNodes(t *testing.T) { + agg, delivered := newTestSigningAggregator(2, time.Hour) + + agg.Add(errorResult("node0", "first")) + if got := len(delivered()); got != 0 { + t.Fatalf("delivered %d results before error threshold", got) + } + agg.Add(errorResult("node1", "second")) + agg.Add(successResult("node2")) + + got := delivered() + if len(got) != 1 || got[0].ErrorReason != "first" { + t.Fatalf("expected the first error delivered once, got %+v", got) + } +} + +func TestResultAggregatorDeliversLoneErrorOnTimeout(t *testing.T) { + agg, delivered := newTestSigningAggregator(2, 50*time.Millisecond) + + agg.Add(errorResult("node0", "boom")) + time.Sleep(100 * time.Millisecond) + + got := delivered() + if len(got) != 1 || got[0].ErrorReason != "boom" { + t.Fatalf("expected the lone error after timeout, got %+v", got) + } +} + +func TestReshareAggregatorUsesPerSessionThreshold(t *testing.T) { + var delivered []event.ResharingResultEvent + agg := reshareResultAggregator(func(e event.ResharingResultEvent) int { + if e.SessionID == "small" { + return 2 + } + return 3 + }, func(e event.ResharingResultEvent) { delivered = append(delivered, e) }) + agg.cfg.timeout = time.Hour + + for _, node := range []string{"node0", "node1"} { + agg.Add(event.ResharingResultEvent{SessionID: "small", WalletID: "w1", ResultType: event.ResultTypeSuccess, NodeID: node}) + agg.Add(event.ResharingResultEvent{SessionID: "large", WalletID: "w1", ResultType: event.ResultTypeSuccess, NodeID: node}) + } + + if len(delivered) != 1 || delivered[0].SessionID != "small" { + t.Fatalf("expected only the small-committee session delivered, got %+v", delivered) + } +} diff --git a/pkg/event/keygen.go b/pkg/event/keygen.go index 78ab6319..40bc99aa 100644 --- a/pkg/event/keygen.go +++ b/pkg/event/keygen.go @@ -14,4 +14,9 @@ type KeygenResultEvent struct { ResultType ResultType `json:"result_type"` ErrorReason string `json:"error_reason"` ErrorCode string `json:"error_code"` + + // NodeID and ResultSignature authenticate the node that emitted this + // result; see SigningResultEvent. + NodeID string `json:"node_id,omitempty"` + ResultSignature []byte `json:"result_signature,omitempty"` } diff --git a/pkg/event/reshare.go b/pkg/event/reshare.go index 36153880..00430727 100644 --- a/pkg/event/reshare.go +++ b/pkg/event/reshare.go @@ -3,6 +3,7 @@ package event import "github.com/fystack/mpcium/pkg/types" type ResharingResultEvent struct { + SessionID string `json:"session_id"` WalletID string `json:"wallet_id"` NewThreshold int `json:"new_threshold"` KeyType types.KeyType `json:"key_type"` @@ -11,4 +12,9 @@ type ResharingResultEvent struct { ResultType ResultType `json:"result_type"` ErrorReason string `json:"error_reason"` ErrorCode string `json:"error_code"` + + // NodeID and ResultSignature authenticate the node that emitted this + // result; see SigningResultEvent. + NodeID string `json:"node_id,omitempty"` + ResultSignature []byte `json:"result_signature,omitempty"` } diff --git a/pkg/event/result.go b/pkg/event/result.go new file mode 100644 index 00000000..fd3dc7ef --- /dev/null +++ b/pkg/event/result.go @@ -0,0 +1,101 @@ +package event + +import ( + "crypto/ed25519" + "encoding/json" + "fmt" +) + +// NodeResult is a result event that a node signs with its identity key so the +// consumer can verify it was produced by the MPC network and not injected or +// tampered with by a compromised message broker. +type nodeResult interface { + // Raw returns the canonical bytes that the node signed (signature cleared). + Raw() ([]byte, error) + setNodeSignature(nodeID string, signature []byte) +} + +// ResultSigner is the node's identity key holder (identity.Store satisfies it). +type ResultSigner interface { + NodeID() string + Sign(data []byte) ([]byte, error) +} + +// NodeKeyLookup returns the ed25519 public key of an MPC node. +type NodeKeyLookup func(nodeID string) ([]byte, error) + +// SignNodeResult fills NodeID and ResultSignature on a result event. +func SignNodeResult(res nodeResult, signer ResultSigner) error { + if signer == nil { + return fmt.Errorf("result signer is required") + } + raw, err := res.Raw() + if err != nil { + return fmt.Errorf("failed to marshal result for signing: %w", err) + } + sig, err := signer.Sign(raw) + if err != nil { + return fmt.Errorf("failed to sign result: %w", err) + } + res.setNodeSignature(signer.NodeID(), sig) + return nil +} + +// VerifyNodeResult checks a result event's signature against the emitting +// node's ed25519 public key. It returns an error when the event is unsigned, +// comes from an unknown node, or the signature does not match. +func VerifyNodeResult(nodeID string, signature []byte, res interface{ Raw() ([]byte, error) }, lookup NodeKeyLookup) error { + if lookup == nil { + return nil + } + if nodeID == "" || len(signature) == 0 { + return fmt.Errorf("result event is not signed by the node") + } + publicKey, err := lookup(nodeID) + if err != nil { + return fmt.Errorf("failed to get public key of node %s: %w", nodeID, err) + } + raw, err := res.Raw() + if err != nil { + return fmt.Errorf("failed to marshal result for verification: %w", err) + } + if !ed25519.Verify(publicKey, raw, signature) { + return fmt.Errorf("invalid result signature from node %s", nodeID) + } + return nil +} + +func (e *SigningResultEvent) Raw() ([]byte, error) { + raw := *e + raw.NodeID, raw.ResultSignature = "", nil + return json.Marshal(raw) +} + +func (e *SigningResultEvent) setNodeSignature(nodeID string, signature []byte) { + e.NodeID, e.ResultSignature = nodeID, signature +} + +func (e *KeygenResultEvent) Raw() ([]byte, error) { + raw := *e + raw.NodeID, raw.ResultSignature = "", nil + return json.Marshal(raw) +} + +func (e *KeygenResultEvent) setNodeSignature(nodeID string, signature []byte) { + e.NodeID, e.ResultSignature = nodeID, signature +} + +func (e *ResharingResultEvent) Raw() ([]byte, error) { + raw := *e + raw.NodeID, raw.ResultSignature = "", nil + return json.Marshal(raw) +} + +func (e *ResharingResultEvent) setNodeSignature(nodeID string, signature []byte) { + e.NodeID, e.ResultSignature = nodeID, signature +} + +// NodeScopedIdempotentKey keeps JetStream from deduplicating the same result published by different nodes. +func NodeScopedIdempotentKey(key, nodeID string) string { + return key + ":" + nodeID +} diff --git a/pkg/event/result_test.go b/pkg/event/result_test.go new file mode 100644 index 00000000..05696eea --- /dev/null +++ b/pkg/event/result_test.go @@ -0,0 +1,87 @@ +package event + +import ( + "crypto/ed25519" + "crypto/rand" + "fmt" + "testing" + + "github.com/stretchr/testify/require" +) + +type testSigner struct { + nodeID string + priv ed25519.PrivateKey +} + +func (s *testSigner) NodeID() string { return s.nodeID } +func (s *testSigner) Sign(data []byte) ([]byte, error) { return ed25519.Sign(s.priv, data), nil } + +func newTestSigner(t *testing.T) *testSigner { + t.Helper() + _, priv, err := ed25519.GenerateKey(rand.Reader) + require.NoError(t, err) + return &testSigner{nodeID: "node0", priv: priv} +} + +func newTestLookup(t *testing.T, signer *testSigner) NodeKeyLookup { + t.Helper() + pub, ok := signer.priv.Public().(ed25519.PublicKey) + require.True(t, ok) + return func(nodeID string) ([]byte, error) { + if nodeID != signer.nodeID { + return nil, fmt.Errorf("unknown node: %s", nodeID) + } + return pub, nil + } +} + +func TestSignAndVerifyNodeResult(t *testing.T) { + signer := newTestSigner(t) + lookup := newTestLookup(t, signer) + + t.Run("signing result", func(t *testing.T) { + e := SigningResultEvent{WalletID: "w1", TxID: "tx1", R: []byte{1}, S: []byte{2}} + require.NoError(t, SignNodeResult(&e, signer)) + require.Equal(t, signer.nodeID, e.NodeID) + require.NotEmpty(t, e.ResultSignature) + + require.NoError(t, VerifyNodeResult(e.NodeID, e.ResultSignature, &e, lookup)) + + e.WalletID = "tampered" + require.Error(t, VerifyNodeResult(e.NodeID, e.ResultSignature, &e, lookup)) + }) + + t.Run("keygen result", func(t *testing.T) { + e := KeygenResultEvent{WalletID: "w1", ECDSAPubKey: []byte{3}} + require.NoError(t, SignNodeResult(&e, signer)) + require.NoError(t, VerifyNodeResult(e.NodeID, e.ResultSignature, &e, lookup)) + }) + + t.Run("resharing result", func(t *testing.T) { + e := ResharingResultEvent{WalletID: "w1", NewThreshold: 2} + require.NoError(t, SignNodeResult(&e, signer)) + require.NoError(t, VerifyNodeResult(e.NodeID, e.ResultSignature, &e, lookup)) + }) +} + +func TestVerifyNodeResultRejects(t *testing.T) { + signer := newTestSigner(t) + lookup := newTestLookup(t, signer) + + t.Run("unsigned event", func(t *testing.T) { + e := SigningResultEvent{WalletID: "w1"} + require.Error(t, VerifyNodeResult(e.NodeID, e.ResultSignature, &e, lookup)) + }) + + t.Run("unknown node", func(t *testing.T) { + e := SigningResultEvent{WalletID: "w1"} + require.NoError(t, SignNodeResult(&e, signer)) + require.Error(t, VerifyNodeResult("node9", e.ResultSignature, &e, lookup)) + }) + + t.Run("nil lookup skips verification", func(t *testing.T) { + e := SigningResultEvent{WalletID: "w1"} + require.NoError(t, VerifyNodeResult("", nil, &e, nil)) + }) +} diff --git a/pkg/event/sign.go b/pkg/event/sign.go index ebbd0775..10b22f97 100644 --- a/pkg/event/sign.go +++ b/pkg/event/sign.go @@ -22,6 +22,12 @@ type SigningResultEvent struct { // TODO: define two separate events for eddsa and ecdsa Signature []byte `json:"signature"` + + // NodeID and ResultSignature identify and authenticate the node that + // emitted this result, so consumers can reject events injected or + // tampered with by a compromised message broker. + NodeID string `json:"node_id,omitempty"` + ResultSignature []byte `json:"result_signature,omitempty"` } type SigningResultSuccessEvent struct { diff --git a/pkg/event/types.go b/pkg/event/types.go index ac26ca94..7d97ae5e 100644 --- a/pkg/event/types.go +++ b/pkg/event/types.go @@ -73,6 +73,9 @@ const ( ErrorCodeMessageDelivery ErrorCode = "ERROR_MESSAGE_DELIVERY" ErrorCodeMaxDeliveryAttempts ErrorCode = "ERROR_MAX_DELIVERY_ATTEMPTS" + // Result aggregation errors + ErrorCodeInsufficientNodeSignatures ErrorCode = "ERROR_INSUFFICIENT_NODE_SIGNATURES" + // Configuration errors ErrorCodeInvalidConfiguration ErrorCode = "ERROR_INVALID_CONFIGURATION" ErrorCodeInvalidThreshold ErrorCode = "ERROR_INVALID_THRESHOLD" diff --git a/pkg/eventconsumer/event_consumer.go b/pkg/eventconsumer/event_consumer.go index 691c712d..e9ddebef 100644 --- a/pkg/eventconsumer/event_consumer.go +++ b/pkg/eventconsumer/event_consumer.go @@ -1,6 +1,7 @@ package eventconsumer import ( + "cmp" "context" "encoding/json" "errors" @@ -247,6 +248,12 @@ func (ec *eventConsumer) handleKeyGenEvent(natMsg *nats.Msg) { return } + if err := event.SignNodeResult(successEvent, ec.identityStore); err != nil { + logger.Error("Failed to sign keygen success event", err, "walletID", walletID) + ec.handleKeygenSessionError(walletID, err, "Failed to sign keygen success event", natMsg) + return + } + payload, err := json.Marshal(successEvent) if err != nil { logger.Error("Failed to marshal keygen success event", err) @@ -258,7 +265,7 @@ func (ec *eventConsumer) handleKeyGenEvent(natMsg *nats.Msg) { if err := ec.genKeyResultQueue.Enqueue( key, payload, - &messaging.EnqueueOptions{IdempotententKey: composeKeygenIdempotentKey(walletID, natMsg)}, + &messaging.EnqueueOptions{IdempotententKey: event.NodeScopedIdempotentKey(composeKeygenIdempotentKey(walletID, natMsg), successEvent.NodeID)}, ); err != nil { logger.Error("Failed to publish key generation success message", err) ec.handleKeygenSessionError(walletID, err, "Failed to publish key generation success message", natMsg) @@ -279,6 +286,11 @@ func (ec *eventConsumer) handleKeygenSessionError(walletID string, err error, co ErrorReason: fullErrMsg, } + if err := event.SignNodeResult(&keygenResult, ec.identityStore); err != nil { + logger.Error("Failed to sign keygen result event", err, "walletID", walletID) + return + } + keygenResultBytes, err := json.Marshal(keygenResult) if err != nil { logger.Error("Failed to marshal keygen result event", err, @@ -289,7 +301,7 @@ func (ec *eventConsumer) handleKeygenSessionError(walletID string, err error, co key := fmt.Sprintf(mpc.TypeGenerateWalletResultFmt, walletID) err = ec.genKeyResultQueue.Enqueue(key, keygenResultBytes, &messaging.EnqueueOptions{ - IdempotententKey: composeKeygenIdempotentKey(walletID, natMsg), + IdempotententKey: event.NodeScopedIdempotentKey(composeKeygenIdempotentKey(walletID, natMsg), keygenResult.NodeID), }) if err != nil { logger.Error("Failed to enqueue keygen result event", err, @@ -527,6 +539,11 @@ func (ec *eventConsumer) handleSigningSessionError(walletID, txID, networkIntern ErrorReason: fullErrMsg, } + if err := event.SignNodeResult(&signingResult, ec.identityStore); err != nil { + logger.Error("Failed to sign signing result event", err, "txID", txID) + return + } + signingResultBytes, err := json.Marshal(signingResult) if err != nil { logger.Error("Failed to marshal signing result event", err, @@ -536,7 +553,7 @@ func (ec *eventConsumer) handleSigningSessionError(walletID, txID, networkIntern return } err = ec.signingResultQueue.Enqueue(event.SigningResultCompleteTopic, signingResultBytes, &messaging.EnqueueOptions{ - IdempotententKey: composeSigningIdempotentKey(txID, natMsg), + IdempotententKey: event.NodeScopedIdempotentKey(composeSigningIdempotentKey(txID, natMsg), signingResult.NodeID), }) if err != nil { logger.Error("Failed to enqueue signing result event", err, @@ -568,12 +585,13 @@ func (ec *eventConsumer) consumeReshareEvent() error { var msg types.ResharingMessage if err := json.Unmarshal(natMsg.Data, &msg); err != nil { logger.Error("Failed to unmarshal resharing message", err) - ec.handleReshareSessionError(msg.WalletID, msg.KeyType, msg.NewThreshold, err, "Failed to unmarshal resharing message", natMsg) + ec.handleReshareSessionError(msg.SessionID, msg.WalletID, msg.KeyType, msg.NewThreshold, err, "Failed to unmarshal resharing message", natMsg) return } if msg.SessionID == "" { ec.handleReshareSessionError( + msg.SessionID, msg.WalletID, msg.KeyType, msg.NewThreshold, @@ -586,7 +604,7 @@ func (ec *eventConsumer) consumeReshareEvent() error { if err := ec.identityStore.VerifyInitiatorMessage(&msg); err != nil { logger.Error("Failed to verify initiator message", err) - ec.handleReshareSessionError(msg.WalletID, msg.KeyType, msg.NewThreshold, err, "Failed to verify initiator message", natMsg) + ec.handleReshareSessionError(msg.SessionID, msg.WalletID, msg.KeyType, msg.NewThreshold, err, "Failed to verify initiator message", natMsg) return } @@ -596,7 +614,7 @@ func (ec *eventConsumer) consumeReshareEvent() error { sessionType, err := sessionTypeFromKeyType(keyType) if err != nil { logger.Error("Failed to get session type", err) - ec.handleReshareSessionError(walletID, keyType, msg.NewThreshold, err, "Failed to get session type", natMsg) + ec.handleReshareSessionError(msg.SessionID, walletID, keyType, msg.NewThreshold, err, "Failed to get session type", natMsg) return } @@ -614,13 +632,13 @@ func (ec *eventConsumer) consumeReshareEvent() error { oldSession, err := createSession(false) if err != nil { logger.Error("Failed to create old reshare session", err, "walletID", walletID) - ec.handleReshareSessionError(walletID, keyType, msg.NewThreshold, err, "Failed to create old reshare session", natMsg) + ec.handleReshareSessionError(msg.SessionID, walletID, keyType, msg.NewThreshold, err, "Failed to create old reshare session", natMsg) return } newSession, err := createSession(true) if err != nil { logger.Error("Failed to create new reshare session", err, "walletID", walletID) - ec.handleReshareSessionError(walletID, keyType, msg.NewThreshold, err, "Failed to create new reshare session", natMsg) + ec.handleReshareSessionError(msg.SessionID, walletID, keyType, msg.NewThreshold, err, "Failed to create new reshare session", natMsg) return } @@ -633,6 +651,7 @@ func (ec *eventConsumer) consumeReshareEvent() error { var wg sync.WaitGroup successEvent := &event.ResharingResultEvent{ + SessionID: msg.SessionID, WalletID: walletID, NewThreshold: msg.NewThreshold, KeyType: msg.KeyType, @@ -642,7 +661,7 @@ func (ec *eventConsumer) consumeReshareEvent() error { if oldSession != nil { err := oldSession.Init() if err != nil { - ec.handleReshareSessionError(walletID, keyType, msg.NewThreshold, err, "Failed to init old reshare session", natMsg) + ec.handleReshareSessionError(msg.SessionID, walletID, keyType, msg.NewThreshold, err, "Failed to init old reshare session", natMsg) return } oldSession.ListenToIncomingMessageAsync() @@ -651,7 +670,7 @@ func (ec *eventConsumer) consumeReshareEvent() error { if newSession != nil { err := newSession.Init() if err != nil { - ec.handleReshareSessionError(walletID, keyType, msg.NewThreshold, err, "Failed to init new reshare session", natMsg) + ec.handleReshareSessionError(msg.SessionID, walletID, keyType, msg.NewThreshold, err, "Failed to init new reshare session", natMsg) return } newSession.ListenToIncomingMessageAsync() @@ -676,7 +695,7 @@ func (ec *eventConsumer) consumeReshareEvent() error { return case err := <-oldSession.ErrChan(): logger.Error("Old reshare session error", err) - ec.handleReshareSessionError(walletID, keyType, msg.NewThreshold, err, "Old reshare session error", natMsg) + ec.handleReshareSessionError(msg.SessionID, walletID, keyType, msg.NewThreshold, err, "Old reshare session error", natMsg) doneOld() return } @@ -697,7 +716,7 @@ func (ec *eventConsumer) consumeReshareEvent() error { return case err := <-newSession.ErrChan(): logger.Error("New reshare session error", err) - ec.handleReshareSessionError(walletID, keyType, msg.NewThreshold, err, "New reshare session error", natMsg) + ec.handleReshareSessionError(msg.SessionID, walletID, keyType, msg.NewThreshold, err, "New reshare session error", natMsg) doneNew() return } @@ -709,10 +728,16 @@ func (ec *eventConsumer) consumeReshareEvent() error { logger.Info("Reshare session finished", "walletID", walletID, "pubKey", fmt.Sprintf("%x", successEvent.PubKey)) if newSession != nil { + if err := event.SignNodeResult(successEvent, ec.identityStore); err != nil { + logger.Error("Failed to sign reshare success event", err, "walletID", walletID) + ec.handleReshareSessionError(msg.SessionID, walletID, keyType, msg.NewThreshold, err, "Failed to sign reshare success event", natMsg) + return + } + successBytes, err := json.Marshal(successEvent) if err != nil { logger.Error("Failed to marshal reshare success event", err) - ec.handleReshareSessionError(walletID, keyType, msg.NewThreshold, err, "Failed to marshal reshare success event", natMsg) + ec.handleReshareSessionError(msg.SessionID, walletID, keyType, msg.NewThreshold, err, "Failed to marshal reshare success event", natMsg) return } @@ -721,11 +746,11 @@ func (ec *eventConsumer) consumeReshareEvent() error { key, successBytes, &messaging.EnqueueOptions{ - IdempotententKey: composeReshareIdempotentKey(msg.SessionID, natMsg), + IdempotententKey: event.NodeScopedIdempotentKey(composeReshareIdempotentKey(msg.SessionID, natMsg), successEvent.NodeID), }) if err != nil { logger.Error("Failed to publish reshare success message", err) - ec.handleReshareSessionError(walletID, keyType, msg.NewThreshold, err, "Failed to publish reshare success message", natMsg) + ec.handleReshareSessionError(msg.SessionID, walletID, keyType, msg.NewThreshold, err, "Failed to publish reshare success message", natMsg) return } logger.Info("[COMPLETED RESHARE] Successfully published", "walletID", walletID) @@ -740,6 +765,7 @@ func (ec *eventConsumer) consumeReshareEvent() error { // handleReshareSessionError handles errors that occur during reshare operations func (ec *eventConsumer) handleReshareSessionError( + sessionID string, walletID string, keyType types.KeyType, newThreshold int, @@ -760,6 +786,7 @@ func (ec *eventConsumer) handleReshareSessionError( ) reshareResult := event.ResharingResultEvent{ + SessionID: sessionID, ResultType: event.ResultTypeError, ErrorCode: string(errorCode), WalletID: walletID, @@ -768,6 +795,11 @@ func (ec *eventConsumer) handleReshareSessionError( ErrorReason: fullErrMsg, } + if err := event.SignNodeResult(&reshareResult, ec.identityStore); err != nil { + logger.Error("Failed to sign reshare result event", err, "walletID", walletID) + return + } + reshareResultBytes, err := json.Marshal(reshareResult) if err != nil { logger.Error("Failed to marshal reshare result event", err, @@ -778,7 +810,7 @@ func (ec *eventConsumer) handleReshareSessionError( key := fmt.Sprintf(mpc.TypeReshareWalletResultFmt, walletID) err = ec.reshareResultQueue.Enqueue(key, reshareResultBytes, &messaging.EnqueueOptions{ - IdempotententKey: composeReshareIdempotentKey(walletID, natMsg), + IdempotententKey: event.NodeScopedIdempotentKey(composeReshareIdempotentKey(cmp.Or(sessionID, walletID), natMsg), reshareResult.NodeID), }) if err != nil { logger.Error("Failed to enqueue reshare result event", err, diff --git a/pkg/eventconsumer/keygen_consumer.go b/pkg/eventconsumer/keygen_consumer.go index 98d9f072..612486d7 100644 --- a/pkg/eventconsumer/keygen_consumer.go +++ b/pkg/eventconsumer/keygen_consumer.go @@ -8,6 +8,7 @@ import ( "time" "github.com/fystack/mpcium/pkg/event" + "github.com/fystack/mpcium/pkg/identity" "github.com/fystack/mpcium/pkg/logger" "github.com/fystack/mpcium/pkg/messaging" "github.com/fystack/mpcium/pkg/mpc" @@ -39,6 +40,7 @@ type keygenConsumer struct { jsBroker messaging.MessageBroker peerRegistry mpc.PeerRegistry keygenResultQueue messaging.MessageQueue + identityStore identity.Store // jsSub holds the JetStream subscription, so it can be cleaned up during Close(). jsSub messaging.MessageSubscription @@ -51,6 +53,7 @@ func NewKeygenConsumer( pubsub messaging.PubSub, peerRegistry mpc.PeerRegistry, keygenResultQueue messaging.MessageQueue, + identityStore identity.Store, ) KeygenConsumer { return &keygenConsumer{ natsConn: natsConn, @@ -58,6 +61,7 @@ func NewKeygenConsumer( jsBroker: jsBroker, peerRegistry: peerRegistry, keygenResultQueue: keygenResultQueue, + identityStore: identityStore, } } @@ -199,6 +203,11 @@ func (sc *keygenConsumer) handleKeygenError(keygenMsg types.GenerateKeyMessage, ErrorReason: err.Error(), } + if signErr := event.SignNodeResult(&keygenResult, sc.identityStore); signErr != nil { + logger.Error("Failed to sign keygen result event", signErr, "walletID", keygenResult.WalletID) + return + } + keygenResultBytes, err := json.Marshal(keygenResult) if err != nil { logger.Error("Failed to marshal keygen result event", err, @@ -209,7 +218,7 @@ func (sc *keygenConsumer) handleKeygenError(keygenMsg types.GenerateKeyMessage, topic := fmt.Sprintf(mpc.TypeGenerateWalletResultFmt, keygenResult.WalletID) err = sc.keygenResultQueue.Enqueue(topic, keygenResultBytes, &messaging.EnqueueOptions{ - IdempotententKey: buildIdempotentKey(keygenMsg.WalletID, sessionID, mpc.TypeGenerateWalletResultFmt), + IdempotententKey: event.NodeScopedIdempotentKey(buildIdempotentKey(keygenMsg.WalletID, sessionID, mpc.TypeGenerateWalletResultFmt), keygenResult.NodeID), }) if err != nil { logger.Error("Failed to enqueue keygen result event", err, diff --git a/pkg/eventconsumer/sign_consumer.go b/pkg/eventconsumer/sign_consumer.go index 24f21fec..c40937a1 100644 --- a/pkg/eventconsumer/sign_consumer.go +++ b/pkg/eventconsumer/sign_consumer.go @@ -7,6 +7,7 @@ import ( "time" "github.com/fystack/mpcium/pkg/event" + "github.com/fystack/mpcium/pkg/identity" "github.com/fystack/mpcium/pkg/logger" "github.com/fystack/mpcium/pkg/messaging" "github.com/fystack/mpcium/pkg/mpc" @@ -42,13 +43,14 @@ type signingConsumer struct { peerRegistry mpc.PeerRegistry mpcThreshold int signingResultQueue messaging.MessageQueue + identityStore identity.Store // jsSub holds the JetStream subscription, so it can be cleaned up during Close(). jsSub messaging.Subscription } // NewSigningConsumer returns a new instance of SigningConsumer. -func NewSigningConsumer(natsConn *nats.Conn, jsBroker messaging.MessageBroker, pubsub messaging.PubSub, peerRegistry mpc.PeerRegistry, signingResultQueue messaging.MessageQueue) SigningConsumer { +func NewSigningConsumer(natsConn *nats.Conn, jsBroker messaging.MessageBroker, pubsub messaging.PubSub, peerRegistry mpc.PeerRegistry, signingResultQueue messaging.MessageQueue, identityStore identity.Store) SigningConsumer { mpcThreshold := viper.GetInt("mpc_threshold") return &signingConsumer{ natsConn: natsConn, @@ -57,6 +59,7 @@ func NewSigningConsumer(natsConn *nats.Conn, jsBroker messaging.MessageBroker, p peerRegistry: peerRegistry, mpcThreshold: mpcThreshold, signingResultQueue: signingResultQueue, + identityStore: identityStore, } } @@ -221,6 +224,14 @@ func (sc *signingConsumer) handleSigningError(signMsg types.SignTxMessage, error ErrorReason: err.Error(), } + if signErr := event.SignNodeResult(&signingResult, sc.identityStore); signErr != nil { + logger.Error("Failed to sign signing result event", signErr, + "walletID", signMsg.WalletID, + "txID", signMsg.TxID, + ) + return + } + signingResultBytes, err := json.Marshal(signingResult) if err != nil { logger.Error("Failed to marshal signing result event", err, @@ -231,7 +242,7 @@ func (sc *signingConsumer) handleSigningError(signMsg types.SignTxMessage, error } err = sc.signingResultQueue.Enqueue(event.SigningResultCompleteTopic, signingResultBytes, &messaging.EnqueueOptions{ - IdempotententKey: buildIdempotentKey(signMsg.TxID, sessionID, mpc.TypeSigningResultFmt), + IdempotententKey: event.NodeScopedIdempotentKey(buildIdempotentKey(signMsg.TxID, sessionID, mpc.TypeSigningResultFmt), signingResult.NodeID), }) if err != nil { logger.Error("Failed to enqueue signing result event", err, diff --git a/pkg/eventconsumer/timeout_consumer.go b/pkg/eventconsumer/timeout_consumer.go index bd911700..8393dd6e 100644 --- a/pkg/eventconsumer/timeout_consumer.go +++ b/pkg/eventconsumer/timeout_consumer.go @@ -4,6 +4,7 @@ import ( "encoding/json" "github.com/fystack/mpcium/pkg/event" + "github.com/fystack/mpcium/pkg/identity" "github.com/fystack/mpcium/pkg/logger" "github.com/fystack/mpcium/pkg/messaging" "github.com/nats-io/nats.go" @@ -13,15 +14,17 @@ import ( const maxDeliveriesExceededSubject = "$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.>" type timeOutConsumer struct { - natsConn *nats.Conn - resultQueue messaging.MessageQueue - advisorySub messaging.Subscription + natsConn *nats.Conn + resultQueue messaging.MessageQueue + identityStore identity.Store + advisorySub messaging.Subscription } -func NewTimeOutConsumer(natsConn *nats.Conn, resultQueue messaging.MessageQueue) *timeOutConsumer { +func NewTimeOutConsumer(natsConn *nats.Conn, resultQueue messaging.MessageQueue, identityStore identity.Store) *timeOutConsumer { return &timeOutConsumer{ - natsConn: natsConn, - resultQueue: resultQueue, + natsConn: natsConn, + resultQueue: resultQueue, + identityStore: identityStore, } } @@ -65,6 +68,11 @@ func (tc *timeOutConsumer) Run() { signErrorResult.IsTimeout = true signErrorResult.ErrorReason = "Signing failed: maximum delivery attempts exceeded" + if err := event.SignNodeResult(&signErrorResult, tc.identityStore); err != nil { + logger.Error("Failed to sign signing result event", err, "txID", signErrorResult.TxID) + return + } + signErrorResultBytes, err := json.Marshal(signErrorResult) if err != nil { logger.Error("Failed to marshal signing result event", err) @@ -72,7 +80,7 @@ func (tc *timeOutConsumer) Run() { } err = tc.resultQueue.Enqueue(event.SigningResultTopic, signErrorResultBytes, &messaging.EnqueueOptions{ - IdempotententKey: signErrorResult.TxID, + IdempotententKey: event.NodeScopedIdempotentKey(signErrorResult.TxID, signErrorResult.NodeID), }) if err != nil { logger.Error("Failed to publish signing result event", err) diff --git a/pkg/identity/identity.go b/pkg/identity/identity.go index 2d2f1eee..15dbaf90 100644 --- a/pkg/identity/identity.go +++ b/pkg/identity/identity.go @@ -36,6 +36,10 @@ type NodeIdentity struct { // Store manages node identities type Store interface { + // NodeID returns the ID of the local node + NodeID() string + // Sign signs arbitrary data with the node's identity key + Sign(data []byte) ([]byte, error) // GetPublicKey retrieves a node's public key by its ID GetPublicKey(nodeID string) ([]byte, error) VerifyInitiatorMessage(msg types.InitiatorMessage) error @@ -65,6 +69,7 @@ type InitiatorKey struct { type fileStore struct { identityDir string currentNodeName string + currentNodeID string // Cache for public keys by node_id publicKeys map[string][]byte @@ -110,6 +115,7 @@ func NewFileStore(identityDir, nodeName string, decrypt bool, agePasswordFile st store := &fileStore{ identityDir: identityDir, currentNodeName: nodeName, + currentNodeID: peers[nodeName], publicKeys: make(map[string][]byte), privateKey: privateKey, initiatorKey: initiatorKey, @@ -375,6 +381,19 @@ func (s *fileStore) CheckSymmetricKeyComplete(desired int) bool { return len(s.symmetricKeys) == desired } +// NodeID returns the ID of the local node +func (s *fileStore) NodeID() string { + return s.currentNodeID +} + +// Sign signs arbitrary data with the node's identity key +func (s *fileStore) Sign(data []byte) ([]byte, error) { + if s.privateKey == nil { + return nil, fmt.Errorf("node private key is not loaded") + } + return ed25519.Sign(s.privateKey, data), nil +} + // GetPublicKey retrieves a node's public key by its ID func (s *fileStore) GetPublicKey(nodeID string) ([]byte, error) { s.mu.RLock() diff --git a/pkg/mpc/ecdsa_signing_session.go b/pkg/mpc/ecdsa_signing_session.go index d8fc3b5e..985b58d6 100644 --- a/pkg/mpc/ecdsa_signing_session.go +++ b/pkg/mpc/ecdsa_signing_session.go @@ -173,6 +173,11 @@ func (s *ecdsaSigningSession) Sign(onSuccess func(data []byte)) { SignatureRecovery: sig.SignatureRecovery, } + if err := event.SignNodeResult(&r, s.identityStore); err != nil { + s.ErrCh <- errors.Wrap(err, "Failed to sign signing result event") + return + } + bytes, err := json.Marshal(r) if err != nil { s.ErrCh <- errors.Wrap(err, "Failed to marshal raw signature") @@ -180,7 +185,7 @@ func (s *ecdsaSigningSession) Sign(onSuccess func(data []byte)) { } err = s.resultQueue.Enqueue(event.SigningResultCompleteTopic, bytes, &messaging.EnqueueOptions{ - IdempotententKey: s.idempotentKey, + IdempotententKey: event.NodeScopedIdempotentKey(s.idempotentKey, r.NodeID), }) if err != nil { s.ErrCh <- errors.Wrap(err, "Failed to publish sign success message") diff --git a/pkg/mpc/eddsa_signing_session.go b/pkg/mpc/eddsa_signing_session.go index d70b242c..93d4be2a 100644 --- a/pkg/mpc/eddsa_signing_session.go +++ b/pkg/mpc/eddsa_signing_session.go @@ -162,6 +162,11 @@ func (s *eddsaSigningSession) Sign(onSuccess func(data []byte)) { Signature: sig.Signature, } + if err := event.SignNodeResult(&r, s.identityStore); err != nil { + s.ErrCh <- errors.Wrap(err, "Failed to sign signing result event") + return + } + bytes, err := json.Marshal(r) if err != nil { s.ErrCh <- errors.Wrap(err, "Failed to marshal raw signature") @@ -169,7 +174,7 @@ func (s *eddsaSigningSession) Sign(onSuccess func(data []byte)) { } err = s.resultQueue.Enqueue(event.SigningResultCompleteTopic, bytes, &messaging.EnqueueOptions{ - IdempotententKey: s.idempotentKey, + IdempotententKey: event.NodeScopedIdempotentKey(s.idempotentKey, r.NodeID), }) if err != nil { s.ErrCh <- errors.Wrap(err, "Failed to publish sign success message")