Implement ECDH for P2P message encryption

.github/workflows/ci.yml

@@ -6,6 +6,9 @@ on:
   pull_request:
     branches: ["*"]
 
+env:
+  GO_VERSION: "1.24"
+
 jobs:
   test:
     runs-on: ubuntu-latest
@@ -15,19 +18,10 @@ jobs:
         uses: actions/checkout@v4
 
       - name: Set up Go
-        uses: actions/setup-go@v4
+        uses: actions/setup-go@v5
         with:
-          go-version: "1.23"
-
-      - name: Cache Go modules
-        uses: actions/cache@v3
-        with:
-          path: |
-            ~/.cache/go-build
-            ~/go/pkg/mod
-          key: ${{ runner.os }}-go-${{ hashFiles('**/go.sum') }}
-          restore-keys: |
-            ${{ runner.os }}-go-
+          go-version: ${{ env.GO_VERSION }}
+          cache: true
 
       - name: Install dependencies
         run: go mod download
@@ -50,9 +44,13 @@ jobs:
         uses: actions/checkout@v4
 
       - name: Set up Go
-        uses: actions/setup-go@v4
+        uses: actions/setup-go@v5
         with:
-          go-version: "1.23"
+          go-version: ${{ env.GO_VERSION }}
+          cache: true
+
+      - name: Clean Go build cache
+        run: go clean -cache -modcache
 
       - name: Run golangci-lint
         uses: golangci/golangci-lint-action@v3
@@ -74,19 +72,10 @@ jobs:
         uses: actions/checkout@v4
 
       - name: Set up Go
-        uses: actions/setup-go@v4
+        uses: actions/setup-go@v5
         with:
-          go-version: "1.23"
-
-      - name: Cache Go modules
-        uses: actions/cache@v3
-        with:
-          path: |
-            ~/.cache/go-build
-            ~/go/pkg/mod
-          key: ${{ runner.os }}-go-${{ hashFiles('**/go.sum') }}
-          restore-keys: |
-            ${{ runner.os }}-go-
+          go-version: ${{ env.GO_VERSION }}
+          cache: true
 
       - name: Install dependencies
         run: go mod download
@@ -153,19 +142,10 @@ jobs:
         uses: actions/checkout@v4
 
       - name: Set up Go
-        uses: actions/setup-go@v4
+        uses: actions/setup-go@v5
         with:
-          go-version: "1.23"
-
-      - name: Cache Go modules
-        uses: actions/cache@v3
-        with:
-          path: |
-            ~/.cache/go-build
-            ~/go/pkg/mod
-          key: ${{ runner.os }}-go-${{ hashFiles('**/go.sum') }}
-          restore-keys: |
-            ${{ runner.os }}-go-
+          go-version: ${{ env.GO_VERSION }}
+          cache: true
 
       - name: Install dependencies
         run: go mod download
@@ -200,19 +180,10 @@ jobs:
         uses: actions/checkout@v4
 
       - name: Set up Go
-        uses: actions/setup-go@v4
-        with:
-          go-version: "1.23"
-
-      - name: Cache Go modules
-        uses: actions/cache@v3
+        uses: actions/setup-go@v5
         with:
-          path: |
-            ~/.cache/go-build
-            ~/go/pkg/mod
-          key: ${{ runner.os }}-go-${{ hashFiles('**/go.sum') }}
-          restore-keys: |
-            ${{ runner.os }}-go-
+          go-version: ${{ env.GO_VERSION }}
+          cache: true
 
       - name: Install dependencies
         run: go mod download
@@ -293,19 +264,10 @@ jobs:
         uses: actions/checkout@v4
 
       - name: Set up Go
-        uses: actions/setup-go@v4
-        with:
-          go-version: "1.23"
-
-      - name: Cache Go modules
-        uses: actions/cache@v3
+        uses: actions/setup-go@v5
         with:
-          path: |
-            ~/.cache/go-build
-            ~/go/pkg/mod
-          key: ${{ runner.os }}-go-${{ hashFiles('**/go.sum') }}
-          restore-keys: |
-            ${{ runner.os }}-go-
+          go-version: ${{ env.GO_VERSION }}
+          cache: true
 
       - name: Build mpcium
         run: go build -v ./cmd/mpcium

.github/workflows/e2e-tests.yml

@@ -7,7 +7,7 @@ on:
     branches: [master]
 
 env:
-  GO_VERSION: "1.23"
+  GO_VERSION: "1.24"
   CGO_ENABLED: 0
   DOCKER_BUILDKIT: 1
   GO_BUILD_FLAGS: -trimpath -ldflags="-s -w"

.gitignore

@@ -17,3 +17,7 @@ e2e/coverage.html
 e2e/logs/
 # Generated config file (template is tracked)
 e2e/config.test.yaml
+node0
+node1
+node2
+config.yaml

cmd/mpcium/main.go

@@ -129,7 +129,6 @@ func runNode(ctx context.Context, c *cli.Command) error {
 	if err != nil {
 		logger.Fatal("Failed to connect to NATS", err)
 	}
-	defer natsConn.Close()
 
 	pubsub := messaging.NewNATSPubSub(natsConn)
 	keygenBroker, err := messaging.NewJetStreamBroker(ctx, natsConn, event.KeygenBrokerStream, []string{
@@ -159,10 +158,10 @@ func runNode(ctx context.Context, c *cli.Command) error {
 	reshareResultQueue := mqManager.NewMessageQueue("mpc_reshare_result")
 	defer reshareResultQueue.Close()
 
-	logger.Info("Node is running", "peerID", nodeID, "name", nodeName)
+	logger.Info("Node is running", "ID", nodeID, "name", nodeName)
 
 	peerNodeIDs := GetPeerIDs(peers)
-	peerRegistry := mpc.NewRegistry(nodeID, peerNodeIDs, consulClient.KV())
+	peerRegistry := mpc.NewRegistry(nodeID, peerNodeIDs, consulClient.KV(), directMessaging, pubsub, identityStore)
 
 	mpcNode := mpc.NewNode(
 		nodeID,
@@ -194,34 +193,46 @@ func runNode(ctx context.Context, c *cli.Command) error {
 
 	timeoutConsumer.Run()
 	defer timeoutConsumer.Close()
-	keygenConsumer := eventconsumer.NewKeygenConsumer(natsConn, keygenBroker, pubsub, peerRegistry)
-	signingConsumer := eventconsumer.NewSigningConsumer(natsConn, signingBroker, pubsub, peerRegistry)
+	keygenConsumer := eventconsumer.NewKeygenConsumer(natsConn, keygenBroker, pubsub, peerRegistry, genKeyResultQueue)
+	signingConsumer := eventconsumer.NewSigningConsumer(natsConn, signingBroker, pubsub, peerRegistry, singingResultQueue)
 
 	// Make the node ready before starting the signing consumer
 	if err := peerRegistry.Ready(); err != nil {
 		logger.Error("Failed to mark peer registry as ready", err)
 	}
 	logger.Info("[READY] Node is ready", "nodeID", nodeID)
+
+	logger.Info("Starting consumers", "nodeID", nodeID)
 	appContext, cancel := context.WithCancel(context.Background())
-	// Setup signal handling to cancel context on termination signals.
+	//Setup signal handling to cancel context on termination signals.
 	go func() {
 		sigChan := make(chan os.Signal, 1)
 		signal.Notify(sigChan, os.Interrupt, syscall.SIGTERM)
 		<-sigChan
 		logger.Warn("Shutdown signal received, canceling context...")
 		cancel()
 
+		// Resign from peer registry first (before closing NATS)
+		if err := peerRegistry.Resign(); err != nil {
+			logger.Error("Failed to resign from peer registry", err)
+		}
+
 		// Gracefully close consumers
 		if err := keygenConsumer.Close(); err != nil {
 			logger.Error("Failed to close keygen consumer", err)
 		}
 		if err := signingConsumer.Close(); err != nil {
 			logger.Error("Failed to close signing consumer", err)
 		}
+
+		err := natsConn.Drain()
+		if err != nil {
+			logger.Error("Failed to drain NATS connection", err)
+		}
 	}()
 
 	var wg sync.WaitGroup
-	errChan := make(chan error, 2)
+	errChan := make(chan error, 3)
 
 	wg.Add(1)
 	go func() {
@@ -250,14 +261,14 @@ func runNode(ctx context.Context, c *cli.Command) error {
 		logger.Info("All consumers have finished")
 		close(errChan)
 	}()
-
 	for err := range errChan {
 		if err != nil {
 			logger.Error("Consumer error received", err)
 			cancel()
 			return err
 		}
 	}
+
 	return nil
 }
 

config.yaml.template

@@ -5,7 +5,7 @@ consul:
 
 mpc_threshold: 2
 environment: development
-badger_password: "your_badger_password"
+badger_password: "F))ysJp?E]ol&I;^"
 event_initiator_pubkey: "event_initiator_pubkey"
 db_path: "."  
 backup_enabled: true

e2e/reshare_test.go

@@ -92,7 +92,7 @@ func testKeyGenerationForResharing(t *testing.T, suite *E2ETestSuite) {
 	require.NoError(t, err, "Failed to setup keygen result listener")
 
 	// Add a small delay to ensure the result listener is fully set up
-	time.Sleep(2 * time.Second)
+	time.Sleep(10 * time.Second)
 
 	// Trigger key generation for all wallets
 	for _, walletID := range walletIDs {
@@ -179,7 +179,7 @@ func testResharingAllNodes(t *testing.T, suite *E2ETestSuite) {
 	require.NoError(t, err, "Failed to setup resharing result listener")
 
 	// Wait for listener setup
-	time.Sleep(2 * time.Second)
+	time.Sleep(10 * time.Second)
 
 	// Test resharing for both key types
 	for i, walletID := range suite.walletIDs {
@@ -360,7 +360,7 @@ func testSigningAfterResharing(t *testing.T, suite *E2ETestSuite) {
 	require.NoError(t, err, "Failed to setup signing result listener")
 
 	// Wait for listener setup
-	time.Sleep(2 * time.Second)
+	time.Sleep(10 * time.Second)
 
 	// Test messages to sign
 	testMessages := []string{

e2e/setup_test_identities.sh

@@ -2,7 +2,6 @@
 
 # E2E Test Identity Setup Script
 # This script sets up identities for testing with separate test database paths
-
 set -e
 
 # Number of test nodes

e2e/sign_test.go

@@ -150,7 +150,7 @@ func testKeyGenerationForSigning(t *testing.T, suite *E2ETestSuite) {
 	require.NoError(t, err, "Failed to setup keygen result listener")
 
 	// Add a small delay to ensure the result listener is fully set up
-	time.Sleep(2 * time.Second)
+	time.Sleep(10 * time.Second)
 
 	// Trigger key generation for all wallets
 	for _, walletID := range walletIDs {

examples/reshare/main.go

@@ -49,10 +49,10 @@ func main() {
 
 	resharingMsg := &types.ResharingMessage{
 		SessionID: uuid.NewString(),
-		WalletID:  "bf2cc849-8e55-47e4-ab73-e17fb1eb690c",
-		NodeIDs:   []string{"d926fa75-72c7-4538-9052-4a064a84981d", "7b1090cd-ffe3-46ff-8375-594dd3204169"}, // new peer IDs
+		WalletID:  "506d2d40-483a-49f1-93c8-27dd4fe9740c",
+		NodeIDs:   []string{"c95c340e-5a18-472d-b9b0-5ac68218213a", "ac37e85f-caca-4bee-8a3a-49a0fe35abff"}, // new peer IDs
 
-		NewThreshold: 2, // t+1 <= len(NodeIDs)
+		NewThreshold: 1, // t+1 <= len(NodeIDs)
 		KeyType:      types.KeyTypeEd25519,
 	}
 	err = mpcClient.Resharing(resharingMsg)

pkg/encryption/aes.go

@@ -4,6 +4,8 @@ import (
 	"crypto/aes"
 	"crypto/cipher"
 	"crypto/rand"
+	"errors"
+	"fmt"
 )
 
 func EncryptAESGCM(plain, key []byte) (ciphertext, nonce []byte, err error) {
@@ -34,3 +36,51 @@ func DecryptAESGCM(ciphertext, key, nonce []byte) ([]byte, error) {
 	}
 	return aead.Open(nil, nonce, ciphertext, nil)
 }
+
+// EncryptAESGCMWithNonceEmbed encrypts plaintext and embeds the nonce at the start of the returned slice.
+func EncryptAESGCMWithNonceEmbed(plaintext, key []byte) ([]byte, error) {
+	block, err := aes.NewCipher(key)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create AES cipher: %w", err)
+	}
+
+	aead, err := cipher.NewGCM(block)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create GCM: %w", err)
+	}
+
+	nonce := make([]byte, aead.NonceSize())
+	if _, err := rand.Read(nonce); err != nil {
+		return nil, fmt.Errorf("failed to generate nonce: %w", err)
+	}
+
+	ciphertext := aead.Seal(nil, nonce, plaintext, nil)
+	return append(nonce, ciphertext...), nil
+}
+
+// DecryptAESGCMWithNonceEmbed decrypts ciphertext where the nonce is embedded at the start of the slice.
+func DecryptAESGCMWithNonceEmbed(data, key []byte) ([]byte, error) {
+	block, err := aes.NewCipher(key)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create AES cipher: %w", err)
+	}
+	aead, err := cipher.NewGCM(block)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create GCM: %w", err)
+	}
+
+	nonceSize := aead.NonceSize()
+	if len(data) < nonceSize {
+		return nil, errors.New("ciphertext too short")
+	}
+
+	nonce := data[:nonceSize]
+	ciphertext := data[nonceSize:]
+
+	plaintext, err := aead.Open(nil, nonce, ciphertext, nil)
+	if err != nil {
+		return nil, fmt.Errorf("decryption failed: %w", err)
+	}
+
+	return plaintext, nil
+}

pkg/event/types.go

@@ -91,6 +91,8 @@ const (
 	// Context and cancellation errors
 	ErrorCodeContextCancelled ErrorCode = "ERROR_CONTEXT_CANCELLED"
 	ErrorCodeOperationAborted ErrorCode = "ERROR_OPERATION_ABORTED"
+	ErrorCodeNotMajority      ErrorCode = "ERROR_NOT_MAJORITY"
+	ErrorCodeClusterNotReady  ErrorCode = "ERROR_CLUSTER_NOT_READY"
 )
 
 // GetErrorCodeFromError attempts to categorize a generic error into a specific error code