From 28a1663913792f8c593c7dfc0b0c4133dd5cdfa1 Mon Sep 17 00:00:00 2001 From: vietddude Date: Thu, 5 Nov 2026 14:42:44 +0700 Subject: [PATCH 1/2] Sign MPC results in NATS headers and verify them on the client Nodes sign the exact bytes they publish (domain || subject || body) with their identity key and send Mpc-Node-Id and Mpc-Result-Sig headers. The message body is unchanged, so existing clients keep working. - pkg/resultauth: sign/verify, key lookup, t+1 aggregator, signing queue - result_signing.per_node_results (default false) keeps one copy per node - client: ResultVerifyMode off|log|enforce, NodePublicKeys, RequiredNodeSignatures(+ per-wallet override); copies are acked only after their result is delivered - mpcium-cli export-node-keys builds node_public_keys.json from identity files - examples, e2e (enforce, t+1), setup script and deployment guides updated --- .gitignore | 3 +- README.md | 19 ++ cmd/mpcium-cli/export-node-keys.go | 84 ++++++ cmd/mpcium-cli/main.go | 30 ++ cmd/mpcium/main.go | 17 +- config.prod.yaml.template | 5 + config.yaml.template | 5 + deployments/ecs/ecs-instruction.md | 11 + deployments/kubernetes/k8s-instruction.md | 9 + deployments/systemd/README.md | 6 + e2e/base_test.go | 21 +- e2e/config.test.yaml.template | 2 + e2e/multi_client_routing_test.go | 9 +- examples/authorizers/generate/main.go | 5 +- examples/authorizers/sign/main.go | 5 +- examples/ckd/main.go | 5 +- examples/generate/kms/main.go | 5 +- examples/generate/main.go | 5 +- examples/hdwallet/ecdsa/main.go | 5 +- examples/hdwallet/eddsa/main.go | 5 +- .../exampleclient/result_verification.go | 50 ++++ examples/reshare/main.go | 5 +- examples/sign/main.go | 5 +- pkg/client/client.go | 157 +++++++--- pkg/client/result_verification.go | 131 ++++++++ pkg/client/result_verification_test.go | 280 ++++++++++++++++++ pkg/event/types.go | 9 +- pkg/identity/identity.go | 8 + pkg/messaging/message_queue.go | 71 +++-- pkg/resultauth/aggregator.go | 219 ++++++++++++++ pkg/resultauth/aggregator_test.go | 213 +++++++++++++ pkg/resultauth/queue.go | 45 +++ pkg/resultauth/resultauth.go | 134 +++++++++ pkg/resultauth/resultauth_test.go | 212 +++++++++++++ pkg/resultauth/testdata/vectors.json | 12 + setup_identities.sh | 4 + 36 files changed, 1718 insertions(+), 93 deletions(-) create mode 100644 cmd/mpcium-cli/export-node-keys.go create mode 100644 examples/internal/exampleclient/result_verification.go create mode 100644 pkg/client/result_verification.go create mode 100644 pkg/client/result_verification_test.go create mode 100644 pkg/resultauth/aggregator.go create mode 100644 pkg/resultauth/aggregator_test.go create mode 100644 pkg/resultauth/queue.go create mode 100644 pkg/resultauth/resultauth.go create mode 100644 pkg/resultauth/resultauth_test.go create mode 100644 pkg/resultauth/testdata/vectors.json diff --git a/.gitignore b/.gitignore index 90d18835..cbf19373 100644 --- a/.gitignore +++ b/.gitignore @@ -24,4 +24,5 @@ node2 config.yaml .vscode .vagrant -.chain_code \ No newline at end of file +.chain_code +node_public_keys.json diff --git a/README.md b/README.md index 5f0baabd..36ebe400 100644 --- a/README.md +++ b/README.md @@ -318,6 +318,25 @@ configs := []client.KMSSignerOptions{ **Note**: AWS KMS only supports P256 (ECDSA) keys, not Ed25519. If you need Ed25519, use the local signer. +#### Verifying node-signed results + +Each node signs the results it publishes with its identity key, in the `Mpc-Node-Id` and `Mpc-Result-Sig` NATS headers. The message body is unchanged. A client pins the node public keys and checks every result before using it: + +```go +mpcClient := client.NewMPCClient(client.Options{ + NatsConn: natsConn, + Signer: localSigner, + ResultVerifyMode: client.ResultVerifyEnforce, // off (default) | log | enforce + NodePublicKeys: nodePublicKeys, // node_id -> hex public key + RequiredNodeSignatures: 3, // mpc_threshold + 1 +}) +``` + +- `setup_identities.sh` writes `node_public_keys.json` for new local setups. For an existing setup or a production cluster, run `mpcium-cli export-node-keys --identity-dir ` against any node's identity directory, which holds every node's `*_identity.json`. +- `enforce` rejects unsigned or tampered results and delivers a result only after `RequiredNodeSignatures` nodes sent the same one. Set `result_signing.per_node_results: true` on the nodes, or only one copy of each result reaches the client. +- `log` reports failed checks but still delivers every result. Use it while rolling out. +- With `RequiredNodeSignatures` above 1, give each client process its own `ClientID`. Processes that share one also share a JetStream consumer, which splits the node copies of a result between them, so none may reach the required count. + ## Test with AWS KMS (LocalStack) For local development and testing with AWS KMS functionality, you can use LocalStack to simulate AWS KMS services. diff --git a/cmd/mpcium-cli/export-node-keys.go b/cmd/mpcium-cli/export-node-keys.go new file mode 100644 index 00000000..8c71765a --- /dev/null +++ b/cmd/mpcium-cli/export-node-keys.go @@ -0,0 +1,84 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "os" + "slices" + + "github.com/fystack/mpcium/pkg/resultauth" + "github.com/urfave/cli/v3" +) + +func exportNodeKeys(ctx context.Context, c *cli.Command) error { + publicKeys := make(map[string]string) + for _, identityDir := range c.StringSlice("identity-dir") { + dirKeys, err := resultauth.LoadPublicKeys(identityDir) + if err != nil { + return err + } + for nodeID, publicKey := range dirKeys { + if existing, ok := publicKeys[nodeID]; ok && existing != publicKey { + return fmt.Errorf("node %s has different public keys in the identity directories", nodeID) + } + publicKeys[nodeID] = publicKey + } + } + if len(publicKeys) == 0 { + return fmt.Errorf("no *_identity.json files found in %v", c.StringSlice("identity-dir")) + } + if _, err := resultauth.NewKeyLookup(publicKeys); err != nil { + return err + } + if err := checkAllPeersHaveKeys(c.String("peers"), publicKeys); err != nil { + return err + } + + data, err := json.MarshalIndent(publicKeys, "", " ") + if err != nil { + return fmt.Errorf("failed to encode node public keys: %w", err) + } + data = append(data, '\n') + + outputPath := c.String("output") + if outputPath == "-" { + _, err := os.Stdout.Write(data) + return err + } + if _, err := os.Stat(outputPath); err == nil && !c.Bool("overwrite") { + return fmt.Errorf("file %s already exists, use --overwrite to replace it", outputPath) + } + if err := os.WriteFile(outputPath, data, 0644); err != nil { + return fmt.Errorf("failed to write %s: %w", outputPath, err) + } + fmt.Printf("Wrote %d node public keys to %s\n", len(publicKeys), outputPath) + return nil +} + +func checkAllPeersHaveKeys(peersPath string, publicKeys map[string]string) error { + data, err := os.ReadFile(peersPath) + if errors.Is(err, os.ErrNotExist) { + return nil + } + if err != nil { + return fmt.Errorf("failed to read %s: %w", peersPath, err) + } + var peers map[string]string + if err := json.Unmarshal(data, &peers); err != nil { + return fmt.Errorf("failed to parse %s: %w", peersPath, err) + } + + var missing []string + for name, nodeID := range peers { + if _, ok := publicKeys[nodeID]; !ok { + missing = append(missing, name) + } + } + if len(missing) > 0 { + slices.Sort(missing) + return fmt.Errorf("no identity file found for peers %v", missing) + } + return nil +} diff --git a/cmd/mpcium-cli/main.go b/cmd/mpcium-cli/main.go index a334997d..b2988012 100644 --- a/cmd/mpcium-cli/main.go +++ b/cmd/mpcium-cli/main.go @@ -102,6 +102,36 @@ func main() { }, Action: generateIdentity, }, + { + Name: "export-node-keys", + Usage: "Write node_id -> public key JSON from node identity files, for clients that verify node-signed results", + Flags: []cli.Flag{ + &cli.StringSliceFlag{ + Name: "identity-dir", + Aliases: []string{"i"}, + Value: []string{"identity"}, + Usage: "Directory with _identity.json files; repeat for several directories", + }, + &cli.StringFlag{ + Name: "peers", + Aliases: []string{"p"}, + Value: "peers.json", + Usage: "peers.json used to check that every node has a key; skipped if the file does not exist", + }, + &cli.StringFlag{ + Name: "output", + Aliases: []string{"o"}, + Value: "node_public_keys.json", + Usage: "Output file, or - for stdout", + }, + &cli.BoolFlag{ + Name: "overwrite", + Aliases: []string{"f"}, + Usage: "Overwrite the output file if it already exists", + }, + }, + Action: exportNodeKeys, + }, { Name: "generate-initiator", Usage: "Generate identity files for an event initiator node", diff --git a/cmd/mpcium/main.go b/cmd/mpcium/main.go index b8127ea3..70d52763 100644 --- a/cmd/mpcium/main.go +++ b/cmd/mpcium/main.go @@ -24,6 +24,7 @@ import ( "github.com/fystack/mpcium/pkg/logger" "github.com/fystack/mpcium/pkg/messaging" "github.com/fystack/mpcium/pkg/mpc" + "github.com/fystack/mpcium/pkg/resultauth" "github.com/fystack/mpcium/pkg/security" "github.com/hashicorp/consul/api" "github.com/nats-io/nats.go" @@ -225,11 +226,21 @@ func runNode(ctx context.Context, c *cli.Command) error { directMessaging := messaging.NewNatsDirectMessaging(natsConn) mqManager := messaging.NewNATsMessageQueueManager("mpc", event.ResultStreamSubjects(), natsConn) - genKeyResultQueue := mqManager.NewMessageQueue("mpc_keygen_result", event.KeygenResultSubscriptionSubject("")) + resultQueueOpts := resultauth.QueueOptions{PerNodeResults: viper.GetBool("result_signing.per_node_results")} + genKeyResultQueue := resultauth.NewSigningQueue( + mqManager.NewMessageQueue("mpc_keygen_result", event.KeygenResultSubscriptionSubject("")), + identityStore, nodeID, resultQueueOpts, + ) defer genKeyResultQueue.Close() - singingResultQueue := mqManager.NewMessageQueue("mpc_signing_result", event.SigningResultSubscriptionSubject("")) + singingResultQueue := resultauth.NewSigningQueue( + mqManager.NewMessageQueue("mpc_signing_result", event.SigningResultSubscriptionSubject("")), + identityStore, nodeID, resultQueueOpts, + ) defer singingResultQueue.Close() - reshareResultQueue := mqManager.NewMessageQueue("mpc_reshare_result", event.ReshareResultSubscriptionSubject("")) + reshareResultQueue := resultauth.NewSigningQueue( + mqManager.NewMessageQueue("mpc_reshare_result", event.ReshareResultSubscriptionSubject("")), + identityStore, nodeID, resultQueueOpts, + ) defer reshareResultQueue.Close() logger.Info("Starting mpcium node", "version", Version, "ID", nodeID, "name", nodeName) diff --git a/config.prod.yaml.template b/config.prod.yaml.template index 0f406d4b..4b880317 100644 --- a/config.prod.yaml.template +++ b/config.prod.yaml.template @@ -34,3 +34,8 @@ session_warm_up_delay_ms: 100 healthcheck: enabled: false # disabled by default, set to true for cloud deployment address: "127.0.0.1:8080" + +# Keep every node's signed copy of a result so clients can require t+1 matching results. +# Enable only after all clients verify node signatures. +result_signing: + per_node_results: false diff --git a/config.yaml.template b/config.yaml.template index 83243b7f..cfd08a21 100644 --- a/config.yaml.template +++ b/config.yaml.template @@ -25,6 +25,11 @@ healthcheck: enabled: false # disabled by default, set to true for cloud deployment address: "0.0.0.0:8080" +# Keep every node's signed copy of a result so clients can require t+1 matching results. +# Enable only after all clients verify node signatures. +result_signing: + per_node_results: false + # Authorization (optional) # authorization: diff --git a/deployments/ecs/ecs-instruction.md b/deployments/ecs/ecs-instruction.md index 162499d2..42ce86ee 100644 --- a/deployments/ecs/ecs-instruction.md +++ b/deployments/ecs/ecs-instruction.md @@ -137,6 +137,12 @@ identity/node0/ └── node2_identity.json ``` +Collect the node public keys for MPC clients. Clients pin them to check that results were signed by these nodes. They are public keys, not secrets. For a deployment that is already running, run the same command against any node's identity directory, which holds every node's `*_identity.json`: + +```bash +mpcium-cli export-node-keys --identity-dir identity/node0 +``` + ### Step 5: Generate Chain Code ```bash @@ -660,6 +666,11 @@ backup_period_seconds: 300 healthcheck: enabled: true address: "0.0.0.0:8080" + +# Keep every node's signed copy of a result so clients can require t+1 matching results. +# Enable only after all clients verify node signatures. +result_signing: + per_node_results: false ``` ## Post-Deployment Verification diff --git a/deployments/kubernetes/k8s-instruction.md b/deployments/kubernetes/k8s-instruction.md index ded405d8..68d0ec10 100644 --- a/deployments/kubernetes/k8s-instruction.md +++ b/deployments/kubernetes/k8s-instruction.md @@ -200,6 +200,12 @@ identity/node0/ └── node0_private.key.age (encrypted private key) ``` +Collect the node public keys for MPC clients. Clients pin them to check that results were signed by these nodes. They are public keys, not secrets. For a deployment that is already running, run the same command against any node's identity directory, which holds every node's `*_identity.json`: + +```bash +mpcium-cli export-node-keys --identity-dir identity/node0 --identity-dir identity/node1 --identity-dir identity/node2 +``` + ### Step 5: Prepare Secrets You'll need the following secrets for each node: @@ -303,6 +309,9 @@ data: chain_code: "<64-char-hex-string>" # Required: 32-byte hex for HD key derivation backup_enabled: false backup_period_seconds: 300 + # Enable only after all clients verify node signatures (t+1 needs every node's copy) + result_signing: + per_node_results: false ``` ### Step 5: Register Peers in Consul diff --git a/deployments/systemd/README.md b/deployments/systemd/README.md index bb0180ef..863c2535 100644 --- a/deployments/systemd/README.md +++ b/deployments/systemd/README.md @@ -218,6 +218,12 @@ scp /opt/mpcium/identity/node0_identity.json root@:/opt/mpcium/identit ⚠️ Never copy `*_private.key` / `*_private.key.age` to other nodes. +Collect the node public keys for MPC clients. Clients pin them to check that results were signed by these nodes. They are public keys, not secrets. For a deployment that is already running, run the same command against any node's identity directory, which holds every node's `*_identity.json`: + +```bash +cd /opt/mpcium && mpcium-cli export-node-keys +``` + Before continuing, double-check on every node that: - `/opt/mpcium/peers.json` exists diff --git a/e2e/base_test.go b/e2e/base_test.go index 1e68de25..09be2bb8 100644 --- a/e2e/base_test.go +++ b/e2e/base_test.go @@ -18,6 +18,7 @@ import ( "github.com/fystack/mpcium/pkg/client" "github.com/fystack/mpcium/pkg/event" "github.com/fystack/mpcium/pkg/kvstore" + "github.com/fystack/mpcium/pkg/resultauth" "github.com/fystack/mpcium/pkg/types" "github.com/hashicorp/consul/api" "github.com/nats-io/nats.go" @@ -186,6 +187,17 @@ func (s *E2ETestSuite) setupClients(t *testing.T) { t.Log("Clients setup completed") } +// e2eRequiredNodeSignatures is mpc_threshold + 1 from config.test.yaml.template. +const e2eRequiredNodeSignatures = 2 + +func (s *E2ETestSuite) nodePublicKeys(t *testing.T) map[string]string { + t.Helper() + publicKeys, err := resultauth.LoadPublicKeys(filepath.Join(s.testDir, "test_node0", "identity")) + require.NoError(t, err, "Failed to load node public keys") + require.NotEmpty(t, publicKeys, "No node identity files found") + return publicKeys +} + func (s *E2ETestSuite) SetupMPCClient(t *testing.T) { t.Log("Setting up MPC client...") @@ -207,9 +219,12 @@ func (s *E2ETestSuite) SetupMPCClient(t *testing.T) { } mpcClient := client.NewMPCClient(client.Options{ - NatsConn: s.natsConn, - Signer: localSigner, - ClientID: "e2e-suite", + NatsConn: s.natsConn, + Signer: localSigner, + ClientID: "e2e-suite", + ResultVerifyMode: client.ResultVerifyEnforce, + NodePublicKeys: s.nodePublicKeys(t), + RequiredNodeSignatures: e2eRequiredNodeSignatures, }) s.mpcClient = mpcClient t.Log("MPC client created") diff --git a/e2e/config.test.yaml.template b/e2e/config.test.yaml.template index 910ae6c8..b144feb6 100644 --- a/e2e/config.test.yaml.template +++ b/e2e/config.test.yaml.template @@ -11,6 +11,8 @@ nats: max_concurrent_keygen: 1 max_concurrent_signing: 10 session_warm_up_delay_ms: 500 +result_signing: + per_node_results: true # Chain Code for HD Wallet Child Key Derivation (REQUIRED) # All nodes MUST use the same chain_code value diff --git a/e2e/multi_client_routing_test.go b/e2e/multi_client_routing_test.go index 83a23f5f..53a5c96e 100644 --- a/e2e/multi_client_routing_test.go +++ b/e2e/multi_client_routing_test.go @@ -170,9 +170,12 @@ func newScopedMPCClient(t *testing.T, suite *E2ETestSuite, clientID string) (cli require.NoError(t, err, "Failed to connect scoped client %s to NATS", clientID) mpcClient := client.NewMPCClient(client.Options{ - NatsConn: natsConn, - Signer: signer, - ClientID: clientID, + NatsConn: natsConn, + Signer: signer, + ClientID: clientID, + ResultVerifyMode: client.ResultVerifyEnforce, + NodePublicKeys: suite.nodePublicKeys(t), + RequiredNodeSignatures: e2eRequiredNodeSignatures, }) return mpcClient, natsConn diff --git a/examples/authorizers/generate/main.go b/examples/authorizers/generate/main.go index e573be48..e2a641c2 100644 --- a/examples/authorizers/generate/main.go +++ b/examples/authorizers/generate/main.go @@ -13,6 +13,7 @@ import ( "syscall" "time" + "github.com/fystack/mpcium/examples/internal/exampleclient" "github.com/fystack/mpcium/pkg/client" "github.com/fystack/mpcium/pkg/config" "github.com/fystack/mpcium/pkg/event" @@ -86,11 +87,11 @@ func main() { authorizerSigners[authorizerID] = signer } - mpcClient := client.NewMPCClient(client.Options{ + mpcClient := client.NewMPCClient(exampleclient.WithResultVerification(client.Options{ NatsConn: natsConn, Signer: localSigner, ClientID: *clientID, - }) + })) var walletStartTimes sync.Map var walletIDs []string diff --git a/examples/authorizers/sign/main.go b/examples/authorizers/sign/main.go index 452a2b82..f3d93002 100644 --- a/examples/authorizers/sign/main.go +++ b/examples/authorizers/sign/main.go @@ -9,6 +9,7 @@ import ( "slices" "syscall" + "github.com/fystack/mpcium/examples/internal/exampleclient" "github.com/fystack/mpcium/pkg/client" "github.com/fystack/mpcium/pkg/config" "github.com/fystack/mpcium/pkg/event" @@ -80,11 +81,11 @@ func main() { authorizerSigners[authorizerID] = signer } - mpcClient := client.NewMPCClient(client.Options{ + mpcClient := client.NewMPCClient(exampleclient.WithResultVerification(client.Options{ NatsConn: natsConn, Signer: localSigner, ClientID: *clientID, - }) + })) // Create a signing request with authorizers txID := uuid.New().String() diff --git a/examples/ckd/main.go b/examples/ckd/main.go index 35c70750..a298d369 100644 --- a/examples/ckd/main.go +++ b/examples/ckd/main.go @@ -8,6 +8,7 @@ import ( "slices" "syscall" + "github.com/fystack/mpcium/examples/internal/exampleclient" "github.com/fystack/mpcium/pkg/client" "github.com/fystack/mpcium/pkg/config" "github.com/fystack/mpcium/pkg/event" @@ -63,11 +64,11 @@ func main() { logger.Fatal("Failed to create local signer", err) } - mpcClient := client.NewMPCClient(client.Options{ + mpcClient := client.NewMPCClient(exampleclient.WithResultVerification(client.Options{ NatsConn: natsConn, Signer: localSigner, ClientID: *clientID, - }) + })) // 2) Once wallet exists, immediately fire a SignTransaction txID := uuid.New().String() diff --git a/examples/generate/kms/main.go b/examples/generate/kms/main.go index 78397991..cc8759c2 100644 --- a/examples/generate/kms/main.go +++ b/examples/generate/kms/main.go @@ -11,6 +11,7 @@ import ( "syscall" "time" + "github.com/fystack/mpcium/examples/internal/exampleclient" "github.com/fystack/mpcium/pkg/client" "github.com/fystack/mpcium/pkg/config" "github.com/fystack/mpcium/pkg/event" @@ -62,11 +63,11 @@ func main() { } logger.Info("Public key", "key", pubKey) - mpcClient := client.NewMPCClient(client.Options{ + mpcClient := client.NewMPCClient(exampleclient.WithResultVerification(client.Options{ NatsConn: natsConn, Signer: kmsSigner, ClientID: *clientID, - }) + })) var walletStartTimes sync.Map var walletIDs []string diff --git a/examples/generate/main.go b/examples/generate/main.go index ff2c1770..33f98d47 100644 --- a/examples/generate/main.go +++ b/examples/generate/main.go @@ -12,6 +12,7 @@ import ( "syscall" "time" + "github.com/fystack/mpcium/examples/internal/exampleclient" "github.com/fystack/mpcium/pkg/client" "github.com/fystack/mpcium/pkg/config" "github.com/fystack/mpcium/pkg/event" @@ -69,11 +70,11 @@ func main() { logger.Fatal("Failed to create local signer", err) } - mpcClient := client.NewMPCClient(client.Options{ + mpcClient := client.NewMPCClient(exampleclient.WithResultVerification(client.Options{ NatsConn: natsConn, Signer: localSigner, ClientID: *clientID, - }) + })) var walletStartTimes sync.Map var walletIDs []string diff --git a/examples/hdwallet/ecdsa/main.go b/examples/hdwallet/ecdsa/main.go index 2293dae2..ef4744be 100644 --- a/examples/hdwallet/ecdsa/main.go +++ b/examples/hdwallet/ecdsa/main.go @@ -12,6 +12,7 @@ import ( "github.com/btcsuite/btcd/btcec/v2" "github.com/btcsuite/btcd/btcec/v2/ecdsa" + "github.com/fystack/mpcium/examples/internal/exampleclient" "github.com/fystack/mpcium/pkg/ckdutil" "github.com/fystack/mpcium/pkg/client" "github.com/fystack/mpcium/pkg/config" @@ -92,11 +93,11 @@ func main() { logger.Fatal("Failed to create local signer", err) } - mpcClient := client.NewMPCClient(client.Options{ + mpcClient := client.NewMPCClient(exampleclient.WithResultVerification(client.Options{ NatsConn: natsConn, Signer: localSigner, ClientID: *clientID, - }) + })) // Step 1: Generate ONE master wallet fmt.Println("Step 1: Generating master MPC wallet...") diff --git a/examples/hdwallet/eddsa/main.go b/examples/hdwallet/eddsa/main.go index bbc13edc..9436e661 100644 --- a/examples/hdwallet/eddsa/main.go +++ b/examples/hdwallet/eddsa/main.go @@ -15,6 +15,7 @@ import ( "github.com/bnb-chain/tss-lib/v3/tss" "github.com/btcsuite/btcutil/base58" "github.com/decred/dcrd/dcrec/edwards/v2" + "github.com/fystack/mpcium/examples/internal/exampleclient" "github.com/fystack/mpcium/pkg/ckdutil" "github.com/fystack/mpcium/pkg/client" "github.com/fystack/mpcium/pkg/config" @@ -96,11 +97,11 @@ func main() { logger.Fatal("Failed to create local signer", err) } - mpcClient := client.NewMPCClient(client.Options{ + mpcClient := client.NewMPCClient(exampleclient.WithResultVerification(client.Options{ NatsConn: natsConn, Signer: localSigner, ClientID: *clientID, - }) + })) // Step 1: Generate ONE master wallet fmt.Println("Step 1: Generating master MPC wallet...") diff --git a/examples/internal/exampleclient/result_verification.go b/examples/internal/exampleclient/result_verification.go new file mode 100644 index 00000000..ebb8d9be --- /dev/null +++ b/examples/internal/exampleclient/result_verification.go @@ -0,0 +1,50 @@ +package exampleclient + +import ( + "errors" + "os" + + "github.com/fystack/mpcium/pkg/client" + "github.com/fystack/mpcium/pkg/logger" + "github.com/fystack/mpcium/pkg/resultauth" + "github.com/spf13/viper" +) + +const ( + nodePublicKeysFile = "node_public_keys.json" + // Local setups made before node_public_keys.json existed still have every node's identity file here. + fallbackIdentityDir = "node0/identity" +) + +// WithResultVerification enforces node-signed results when node public keys are available locally. +func WithResultVerification(opts client.Options) client.Options { + publicKeys, err := loadNodePublicKeys() + if err != nil { + logger.Fatal("Failed to load node public keys", err) + } + if len(publicKeys) == 0 { + logger.Warn("No node public keys found; results are not verified. Run: mpcium-cli export-node-keys --identity-dir node0/identity") + return opts + } + + opts.ResultVerifyMode = client.ResultVerifyEnforce + opts.NodePublicKeys = publicKeys + opts.RequiredNodeSignatures = 1 + // Counting t+1 matching results needs every node's copy, which nodes only keep with per_node_results. + if viper.GetBool("result_signing.per_node_results") { + opts.RequiredNodeSignatures = viper.GetInt("mpc_threshold") + 1 + } + logger.Info("Verifying node-signed results", "nodes", len(publicKeys), "requiredNodeSignatures", opts.RequiredNodeSignatures) + return opts +} + +func loadNodePublicKeys() (map[string]string, error) { + publicKeys, err := resultauth.LoadPublicKeysFile(nodePublicKeysFile) + if !errors.Is(err, os.ErrNotExist) { + return publicKeys, err + } + if _, err := os.Stat(fallbackIdentityDir); errors.Is(err, os.ErrNotExist) { + return nil, nil + } + return resultauth.LoadPublicKeys(fallbackIdentityDir) +} diff --git a/examples/reshare/main.go b/examples/reshare/main.go index 229c3588..65d3964d 100644 --- a/examples/reshare/main.go +++ b/examples/reshare/main.go @@ -8,6 +8,7 @@ import ( "slices" "syscall" + "github.com/fystack/mpcium/examples/internal/exampleclient" "github.com/fystack/mpcium/pkg/client" "github.com/fystack/mpcium/pkg/config" "github.com/fystack/mpcium/pkg/event" @@ -63,11 +64,11 @@ func main() { logger.Fatal("Failed to create local signer", err) } - mpcClient := client.NewMPCClient(client.Options{ + mpcClient := client.NewMPCClient(exampleclient.WithResultVerification(client.Options{ NatsConn: natsConn, Signer: localSigner, ClientID: *clientID, - }) + })) // 3) Listen for signing results err = mpcClient.OnResharingResult(func(evt event.ResharingResultEvent) { diff --git a/examples/sign/main.go b/examples/sign/main.go index 053fbaf5..d79fbf14 100644 --- a/examples/sign/main.go +++ b/examples/sign/main.go @@ -8,6 +8,7 @@ import ( "slices" "syscall" + "github.com/fystack/mpcium/examples/internal/exampleclient" "github.com/fystack/mpcium/pkg/client" "github.com/fystack/mpcium/pkg/config" "github.com/fystack/mpcium/pkg/event" @@ -61,11 +62,11 @@ func main() { logger.Fatal("Failed to create signer", err) } - mpcClient := client.NewMPCClient(client.Options{ + mpcClient := client.NewMPCClient(exampleclient.WithResultVerification(client.Options{ NatsConn: natsConn, Signer: signer, ClientID: *clientID, - }) + })) // 2) Once wallet exists, immediately fire a SignTransaction txID := uuid.New().String() diff --git a/pkg/client/client.go b/pkg/client/client.go index 1926e0ec..c204e515 100644 --- a/pkg/client/client.go +++ b/pkg/client/client.go @@ -5,11 +5,13 @@ import ( "encoding/json" "fmt" "strings" + "sync" "github.com/fystack/mpcium/pkg/event" "github.com/fystack/mpcium/pkg/eventconsumer" "github.com/fystack/mpcium/pkg/logger" "github.com/fystack/mpcium/pkg/messaging" + "github.com/fystack/mpcium/pkg/resultauth" "github.com/fystack/mpcium/pkg/types" "github.com/nats-io/nats.go" ) @@ -27,14 +29,21 @@ type MPCClient interface { } type mpcClient struct { - signingBroker messaging.MessageBroker - keygenBroker messaging.MessageBroker - pubsub messaging.PubSub - genKeySuccessQueue messaging.MessageQueue - signResultQueue messaging.MessageQueue - reshareSuccessQueue messaging.MessageQueue - signer Signer - clientID string + signingBroker messaging.MessageBroker + keygenBroker messaging.MessageBroker + pubsub messaging.PubSub + genKeySuccessQueue messaging.MessageQueue + signResultQueue messaging.MessageQueue + reshareSuccessQueue messaging.MessageQueue + signer Signer + clientID string + resultVerifier resultVerifier + requiredNodeSignaturesForWallet func(walletID string) int + keygenAggregator sharedAggregator[event.KeygenResultEvent] + signingAggregator sharedAggregator[event.SigningResultEvent] + reshareAggregator sharedAggregator[event.ResharingResultEvent] + // reshareRequiredSigs maps a reshare result subject to the new committee's t+1. + reshareRequiredSigs sync.Map } // Options defines configuration options for creating a new MPCClient @@ -45,8 +54,20 @@ type Options struct { // Signer for signing messages Signer Signer - // ClientID optionally scopes result routing for this client instance. + // ClientID scopes result routing; give each process its own when RequiredNodeSignatures > 1. ClientID string + + // ResultVerifyMode controls checking of node signatures on results; defaults to off. + ResultVerifyMode ResultVerifyMode + + // NodePublicKeys maps node ID to its hex ed25519 identity public key. + NodePublicKeys map[string]string + + // RequiredNodeSignatures is how many distinct nodes must send the same result (t+1) in enforce mode. + RequiredNodeSignatures int + + // RequiredNodeSignaturesForWallet overrides RequiredNodeSignatures for wallets with their own threshold. + RequiredNodeSignaturesForWallet func(walletID string) int } type clientResultRouting struct { @@ -67,6 +88,10 @@ func NewMPCClient(opts Options) MPCClient { if err := validateClientID(opts.ClientID); err != nil { logger.Fatal("Invalid client ID", err) } + verifier, err := newResultVerifier(opts.ResultVerifyMode, opts.NodePublicKeys, opts.RequiredNodeSignatures) + if err != nil { + logger.Fatal("Invalid result verification options", err) + } // 2) Create the PubSub for both publish & subscribe signingBroker, err := messaging.NewJetStreamBroker( @@ -102,14 +127,16 @@ func NewMPCClient(opts Options) MPCClient { reshareSuccessQueue := manager.NewMessageQueue(routing.reshareConsumerName, routing.reshareSubject) return &mpcClient{ - signingBroker: signingBroker, - keygenBroker: keygenBroker, - pubsub: pubsub, - genKeySuccessQueue: genKeySuccessQueue, - signResultQueue: signResultQueue, - reshareSuccessQueue: reshareSuccessQueue, - signer: opts.Signer, - clientID: opts.ClientID, + signingBroker: signingBroker, + keygenBroker: keygenBroker, + pubsub: pubsub, + genKeySuccessQueue: genKeySuccessQueue, + signResultQueue: signResultQueue, + reshareSuccessQueue: reshareSuccessQueue, + signer: opts.Signer, + clientID: opts.ClientID, + resultVerifier: verifier, + requiredNodeSignaturesForWallet: opts.RequiredNodeSignaturesForWallet, } } @@ -149,14 +176,21 @@ func (c *mpcClient) CreateWalletWithAuthorizers(walletID string, authorizerSigna // The callback will be invoked whenever a wallet creation result is received. func (c *mpcClient) OnWalletCreationResult(callback func(event event.KeygenResultEvent)) error { - err := c.genKeySuccessQueue.Dequeue(event.KeygenResultSubscriptionSubject(c.clientID), func(msg []byte) error { - var event event.KeygenResultEvent - err := json.Unmarshal(msg, &event) - if err != nil { - return err + aggregator := c.keygenAggregator.register(callback, resultauth.AggregatorConfig[event.KeygenResultEvent]{ + Insufficient: insufficientKeygenResult, + }) + err := c.genKeySuccessQueue.DequeueMessageAsync(event.KeygenResultSubscriptionSubject(c.clientID), func(msg messaging.Message, done func(error)) { + var result event.KeygenResultEvent + if err := json.Unmarshal(msg.Data, &result); err != nil { + done(err) + return } - callback(event) - return nil + handleVerifiedResult(&c.resultVerifier, aggregator, callback, msg, done, resultauth.Result[event.KeygenResultEvent]{ + Key: msg.Subject, + Required: c.requiredFor(result.WalletID), + Success: result.ResultType == event.ResultTypeSuccess, + Event: result, + }) }) if err != nil { @@ -191,14 +225,21 @@ func (c *mpcClient) SignTransaction(msg *types.SignTxMessage) error { } func (c *mpcClient) OnSignResult(callback func(event event.SigningResultEvent)) error { - err := c.signResultQueue.Dequeue(event.SigningResultSubscriptionSubject(c.clientID), func(msg []byte) error { - var event event.SigningResultEvent - err := json.Unmarshal(msg, &event) - if err != nil { - return err + aggregator := c.signingAggregator.register(callback, resultauth.AggregatorConfig[event.SigningResultEvent]{ + Insufficient: insufficientSigningResult, + }) + err := c.signResultQueue.DequeueMessageAsync(event.SigningResultSubscriptionSubject(c.clientID), func(msg messaging.Message, done func(error)) { + var result event.SigningResultEvent + if err := json.Unmarshal(msg.Data, &result); err != nil { + done(err) + return } - callback(event) - return nil + handleVerifiedResult(&c.resultVerifier, aggregator, callback, msg, done, resultauth.Result[event.SigningResultEvent]{ + Key: msg.Subject + ":" + result.TxID, + Required: c.requiredFor(result.WalletID), + Success: result.ResultType == event.ResultTypeSuccess, + Event: result, + }) }) if err != nil { @@ -225,24 +266,41 @@ func (c *mpcClient) Resharing(msg *types.ResharingMessage) error { return fmt.Errorf("Resharing: marshal error: %w", err) } + resultSubject := event.ReshareResultSubject(c.clientID, msg.SessionID) + if c.resultVerifier.required > 1 { + if msg.NewThreshold+1 > c.resultVerifier.nodeCount { + return fmt.Errorf("Resharing: new threshold %d needs %d node signatures but only %d node public keys are configured", + msg.NewThreshold, msg.NewThreshold+1, c.resultVerifier.nodeCount) + } + c.reshareRequiredSigs.Store(resultSubject, msg.NewThreshold+1) + } + if err := c.pubsub.Publish(eventconsumer.MPCReshareEvent, bytes, c.requestHeaders()); err != nil { + c.reshareRequiredSigs.Delete(resultSubject) return fmt.Errorf("Resharing: publish error: %w", err) } return nil } func (c *mpcClient) OnResharingResult(callback func(event event.ResharingResultEvent)) error { - err := c.reshareSuccessQueue.Dequeue(event.ReshareResultSubscriptionSubject(c.clientID), func(msg []byte) error { - logger.Info("Received reshare success message", "raw", string(msg)) - var event event.ResharingResultEvent - err := json.Unmarshal(msg, &event) - if err != nil { - logger.Error("Failed to unmarshal reshare success event", err, "raw", string(msg)) - return err + aggregator := c.reshareAggregator.register(callback, resultauth.AggregatorConfig[event.ResharingResultEvent]{ + Insufficient: insufficientReshareResult, + Completed: func(subject string) { c.reshareRequiredSigs.Delete(subject) }, + }) + err := c.reshareSuccessQueue.DequeueMessageAsync(event.ReshareResultSubscriptionSubject(c.clientID), func(msg messaging.Message, done func(error)) { + logger.Info("Received reshare result message", "raw", string(msg.Data)) + var result event.ResharingResultEvent + if err := json.Unmarshal(msg.Data, &result); err != nil { + logger.Error("Failed to unmarshal reshare result event", err, "raw", string(msg.Data)) + done(err) + return } - logger.Info("Deserialized reshare success event", "event", event) - callback(event) - return nil + handleVerifiedResult(&c.resultVerifier, aggregator, callback, msg, done, resultauth.Result[event.ResharingResultEvent]{ + Key: msg.Subject, + Required: c.reshareRequiredFor(msg.Subject, result.WalletID), + Success: result.ResultType == event.ResultTypeSuccess, + Event: result, + }) }) if err != nil { @@ -252,6 +310,23 @@ func (c *mpcClient) OnResharingResult(callback func(event event.ResharingResultE return nil } +func (c *mpcClient) requiredFor(walletID string) int { + if c.requiredNodeSignaturesForWallet != nil && c.resultVerifier.mode == ResultVerifyEnforce { + if required := c.requiredNodeSignaturesForWallet(walletID); required > 0 { + return required + } + } + return c.resultVerifier.required +} + +// reshareRequiredFor falls back to the wallet's count when the reshare was not sent by this process. +func (c *mpcClient) reshareRequiredFor(subject, walletID string) int { + if stored, ok := c.reshareRequiredSigs.Load(subject); ok { + return stored.(int) + } + return c.requiredFor(walletID) +} + func (c *mpcClient) requestHeaders() map[string]string { if c.clientID == "" { return nil diff --git a/pkg/client/result_verification.go b/pkg/client/result_verification.go new file mode 100644 index 00000000..3371456d --- /dev/null +++ b/pkg/client/result_verification.go @@ -0,0 +1,131 @@ +package client + +import ( + "fmt" + "sync" + "sync/atomic" + + "github.com/fystack/mpcium/pkg/event" + "github.com/fystack/mpcium/pkg/logger" + "github.com/fystack/mpcium/pkg/messaging" + "github.com/fystack/mpcium/pkg/resultauth" +) + +type ResultVerifyMode string + +const ( + ResultVerifyOff ResultVerifyMode = "off" + // ResultVerifyLog checks signatures and logs failures but still delivers every result. + ResultVerifyLog ResultVerifyMode = "log" + // ResultVerifyEnforce rejects unsigned or invalid results and waits for RequiredNodeSignatures matching results. + ResultVerifyEnforce ResultVerifyMode = "enforce" +) + +type resultVerifier struct { + mode ResultVerifyMode + lookup resultauth.KeyLookup + required int + nodeCount int +} + +func newResultVerifier(mode ResultVerifyMode, publicKeys map[string]string, required int) (resultVerifier, error) { + switch mode { + case "", ResultVerifyOff: + return resultVerifier{mode: ResultVerifyOff, required: 1}, nil + case ResultVerifyLog, ResultVerifyEnforce: + default: + return resultVerifier{}, fmt.Errorf("unknown result verify mode %q", mode) + } + + if len(publicKeys) == 0 { + return resultVerifier{}, fmt.Errorf("node public keys are required when result verify mode is %q", mode) + } + if required > len(publicKeys) { + return resultVerifier{}, fmt.Errorf("required node signatures (%d) exceeds the number of node public keys (%d)", required, len(publicKeys)) + } + lookup, err := resultauth.NewKeyLookup(publicKeys) + if err != nil { + return resultVerifier{}, err + } + if mode != ResultVerifyEnforce || required < 1 { + required = 1 + } + return resultVerifier{mode: mode, lookup: lookup, required: required, nodeCount: len(publicKeys)}, nil +} + +func handleVerifiedResult[T any]( + verifier *resultVerifier, + aggregator *resultauth.Aggregator[T], + deliver func(T), + msg messaging.Message, + done func(error), + result resultauth.Result[T], +) { + if verifier.mode != ResultVerifyLog && verifier.mode != ResultVerifyEnforce { + deliver(result.Event) + done(nil) + return + } + + nodeID, err := resultauth.Verify(verifier.lookup, msg.Subject, msg.Data, msg.Header.Get) + if err != nil { + if verifier.mode == ResultVerifyEnforce { + logger.Error("Rejecting result without a valid node signature", err, "subject", msg.Subject, "nodeID", nodeID) + done(messaging.ErrPermament) + return + } + logger.Warn("Result failed node signature verification", "subject", msg.Subject, "nodeID", nodeID, "error", err) + } + if verifier.mode == ResultVerifyLog { + result.Required = 1 + } + + result.NodeID = nodeID + result.Body = msg.Data + result.Ack = func() { done(nil) } + aggregator.Add(result) +} + +func insufficientNodeSignaturesReason(got, required int) string { + return fmt.Sprintf("result was signed by %d node(s) but %d are required; possible node failure or compromise", got, required) +} + +func insufficientKeygenResult(result event.KeygenResultEvent, got, required int) event.KeygenResultEvent { + result.ResultType = event.ResultTypeError + result.ECDSAPubKey, result.EDDSAPubKey = nil, nil + result.ErrorCode = string(event.ErrorCodeInsufficientNodeSignatures) + result.ErrorReason = insufficientNodeSignaturesReason(got, required) + return result +} + +func insufficientSigningResult(result event.SigningResultEvent, got, required int) event.SigningResultEvent { + result.ResultType = event.ResultTypeError + result.R, result.S, result.SignatureRecovery, result.Signature = nil, nil, nil, nil + result.ErrorCode = event.ErrorCodeInsufficientNodeSignatures + result.ErrorReason = insufficientNodeSignaturesReason(got, required) + return result +} + +func insufficientReshareResult(result event.ResharingResultEvent, got, required int) event.ResharingResultEvent { + result.ResultType = event.ResultTypeError + result.PubKey = nil + result.ErrorCode = string(event.ErrorCodeInsufficientNodeSignatures) + result.ErrorReason = insufficientNodeSignaturesReason(got, required) + return result +} + +// sharedAggregator spans subscriptions because JetStream splits a consumer's node copies across them; the latest callback wins. +type sharedAggregator[T any] struct { + once sync.Once + aggregator *resultauth.Aggregator[T] + callback atomic.Pointer[func(T)] +} + +func (s *sharedAggregator[T]) register(callback func(T), cfg resultauth.AggregatorConfig[T]) *resultauth.Aggregator[T] { + s.callback.Store(&callback) + s.once.Do(func() { + cfg.Deliver = func(result T) { (*s.callback.Load())(result) } + s.aggregator = resultauth.NewAggregator(cfg) + }) + return s.aggregator +} diff --git a/pkg/client/result_verification_test.go b/pkg/client/result_verification_test.go new file mode 100644 index 00000000..624002f6 --- /dev/null +++ b/pkg/client/result_verification_test.go @@ -0,0 +1,280 @@ +package client + +import ( + "crypto/ed25519" + "crypto/rand" + "encoding/hex" + "encoding/json" + "testing" + + "github.com/fystack/mpcium/pkg/event" + "github.com/fystack/mpcium/pkg/messaging" + "github.com/fystack/mpcium/pkg/resultauth" + "github.com/fystack/mpcium/pkg/types" + "github.com/nats-io/nats.go" + "github.com/stretchr/testify/require" +) + +type capturingQueue struct { + messaging.MessageQueue + handler func(msg messaging.Message) error + handlers []func(msg messaging.Message) error + settled []error +} + +// DequeueMessageAsync exposes each subscription as a handler that returns the settle error, or nil while still pending. +func (q *capturingQueue) DequeueMessageAsync(_ string, handler func(msg messaging.Message, done func(err error))) error { + q.handler = func(msg messaging.Message) error { + var settleErr error + handler(msg, func(err error) { + settleErr = err + q.settled = append(q.settled, err) + }) + return settleErr + } + q.handlers = append(q.handlers, q.handler) + return nil +} + +type testNode struct { + id string + privateKey ed25519.PrivateKey +} + +func (n testNode) Sign(data []byte) ([]byte, error) { return ed25519.Sign(n.privateKey, data), nil } + +func newTestNodes(t *testing.T, count int) ([]testNode, map[string]string) { + t.Helper() + nodes := make([]testNode, 0, count) + publicKeys := make(map[string]string, count) + for i := range count { + publicKey, privateKey, err := ed25519.GenerateKey(rand.Reader) + require.NoError(t, err) + node := testNode{id: string(rune('a' + i)), privateKey: privateKey} + nodes = append(nodes, node) + publicKeys[node.id] = hex.EncodeToString(publicKey) + } + return nodes, publicKeys +} + +func signedMessage(t *testing.T, node testNode, subject string, payload any) messaging.Message { + t.Helper() + body, err := json.Marshal(payload) + require.NoError(t, err) + headers, err := resultauth.SignHeaders(node, node.id, subject, body) + require.NoError(t, err) + header := nats.Header{} + for k, v := range headers { + header.Set(k, v) + } + return messaging.Message{Subject: subject, Data: body, Header: header} +} + +func newSigningResultClient(t *testing.T, mode ResultVerifyMode, publicKeys map[string]string, required int) (*capturingQueue, *[]event.SigningResultEvent) { + t.Helper() + verifier, err := newResultVerifier(mode, publicKeys, required) + require.NoError(t, err) + queue := &capturingQueue{} + c := &mpcClient{signResultQueue: queue, resultVerifier: verifier} + var delivered []event.SigningResultEvent + require.NoError(t, c.OnSignResult(func(e event.SigningResultEvent) { delivered = append(delivered, e) })) + return queue, &delivered +} + +func TestOnSignResultEnforceWaitsForRequiredNodes(t *testing.T) { + nodes, publicKeys := newTestNodes(t, 3) + queue, delivered := newSigningResultClient(t, ResultVerifyEnforce, publicKeys, 2) + subject := event.SigningResultSubject("") + success := event.SigningResultEvent{ResultType: event.ResultTypeSuccess, TxID: "tx-1", Signature: []byte("sig")} + + require.NoError(t, queue.handler(signedMessage(t, nodes[0], subject, success))) + require.Empty(t, *delivered) + require.NoError(t, queue.handler(signedMessage(t, nodes[1], subject, success))) + require.NoError(t, queue.handler(signedMessage(t, nodes[2], subject, success))) + + require.Len(t, *delivered, 1) + require.Equal(t, "tx-1", (*delivered)[0].TxID) +} + +func TestOnSignResultEnforceRejectsForgedResults(t *testing.T) { + nodes, publicKeys := newTestNodes(t, 3) + queue, delivered := newSigningResultClient(t, ResultVerifyEnforce, publicKeys, 2) + subject := event.SigningResultSubject("") + success := event.SigningResultEvent{ResultType: event.ResultTypeSuccess, TxID: "tx-1", Signature: []byte("sig")} + + unsigned := signedMessage(t, nodes[0], subject, success) + unsigned.Header = nats.Header{} + require.ErrorIs(t, queue.handler(unsigned), messaging.ErrPermament) + + tampered := signedMessage(t, nodes[0], subject, success) + tampered.Data = []byte(`{"result_type":"success","tx_id":"tx-1","signature":"Zm9yZ2Vk"}`) + require.ErrorIs(t, queue.handler(tampered), messaging.ErrPermament) + + _, outsiderKey, err := ed25519.GenerateKey(rand.Reader) + require.NoError(t, err) + outsider := testNode{id: "outsider", privateKey: outsiderKey} + require.ErrorIs(t, queue.handler(signedMessage(t, outsider, subject, success)), messaging.ErrPermament) + + impostor := testNode{id: nodes[1].id, privateKey: outsiderKey} + require.ErrorIs(t, queue.handler(signedMessage(t, impostor, subject, success)), messaging.ErrPermament) + + require.Empty(t, *delivered) +} + +func TestOnSignResultAcksOnlyAfterDelivery(t *testing.T) { + nodes, publicKeys := newTestNodes(t, 3) + queue, delivered := newSigningResultClient(t, ResultVerifyEnforce, publicKeys, 2) + subject := event.SigningResultSubject("") + success := event.SigningResultEvent{ResultType: event.ResultTypeSuccess, TxID: "tx-1", Signature: []byte("sig")} + + require.NoError(t, queue.handler(signedMessage(t, nodes[0], subject, success))) + require.Empty(t, queue.settled, "a copy waiting for quorum must stay unacked so a crash redelivers it") + require.NoError(t, queue.handler(signedMessage(t, nodes[1], subject, success))) + require.Len(t, *delivered, 1) + require.Equal(t, []error{nil, nil}, queue.settled) + + require.NoError(t, queue.handler(signedMessage(t, nodes[2], subject, success))) + require.Len(t, *delivered, 1) + require.Len(t, queue.settled, 3) +} + +func TestOnSignResultLogModeDeliversOnceWithPerNodeCopies(t *testing.T) { + nodes, publicKeys := newTestNodes(t, 3) + queue, delivered := newSigningResultClient(t, ResultVerifyLog, publicKeys, 2) + subject := event.SigningResultSubject("") + success := event.SigningResultEvent{ResultType: event.ResultTypeSuccess, TxID: "tx-1", Signature: []byte("sig")} + + for _, node := range nodes { + require.NoError(t, queue.handler(signedMessage(t, node, subject, success))) + } + require.Len(t, *delivered, 1) + require.Len(t, queue.settled, 3) +} + +func TestOnSignResultUsesWalletThreshold(t *testing.T) { + nodes, publicKeys := newTestNodes(t, 3) + verifier, err := newResultVerifier(ResultVerifyEnforce, publicKeys, 3) + require.NoError(t, err) + queue := &capturingQueue{} + c := &mpcClient{ + signResultQueue: queue, + resultVerifier: verifier, + requiredNodeSignaturesForWallet: func(walletID string) int { return map[string]int{"small-wallet": 2}[walletID] }, + } + var delivered []event.SigningResultEvent + require.NoError(t, c.OnSignResult(func(e event.SigningResultEvent) { delivered = append(delivered, e) })) + + subject := event.SigningResultSubject("") + success := event.SigningResultEvent{ResultType: event.ResultTypeSuccess, WalletID: "small-wallet", TxID: "tx-1", Signature: []byte("sig")} + require.NoError(t, queue.handler(signedMessage(t, nodes[0], subject, success))) + require.NoError(t, queue.handler(signedMessage(t, nodes[1], subject, success))) + require.Len(t, delivered, 1) +} + +func TestResharingRejectsThresholdAboveNodeKeys(t *testing.T) { + _, publicKeys := newTestNodes(t, 3) + verifier, err := newResultVerifier(ResultVerifyEnforce, publicKeys, 2) + require.NoError(t, err) + c := &mpcClient{resultVerifier: verifier, signer: staticSigner{}, pubsub: &recordingPubSub{}} + + err = c.Resharing(&types.ResharingMessage{SessionID: "session-1", NewThreshold: 3, WalletID: "wallet-1"}) + require.Error(t, err) +} + +func TestOnSignResultSubscriptionsShareOneAggregator(t *testing.T) { + nodes, publicKeys := newTestNodes(t, 3) + verifier, err := newResultVerifier(ResultVerifyEnforce, publicKeys, 2) + require.NoError(t, err) + queue := &capturingQueue{} + c := &mpcClient{signResultQueue: queue, resultVerifier: verifier} + var first, second []event.SigningResultEvent + require.NoError(t, c.OnSignResult(func(e event.SigningResultEvent) { first = append(first, e) })) + require.NoError(t, c.OnSignResult(func(e event.SigningResultEvent) { second = append(second, e) })) + require.Len(t, queue.handlers, 2) + + subject := event.SigningResultSubject("") + success := event.SigningResultEvent{ResultType: event.ResultTypeSuccess, TxID: "tx-1", Signature: []byte("sig")} + require.NoError(t, queue.handlers[0](signedMessage(t, nodes[0], subject, success))) + require.NoError(t, queue.handlers[1](signedMessage(t, nodes[1], subject, success))) + require.NoError(t, queue.handlers[0](signedMessage(t, nodes[2], subject, success))) + + require.Empty(t, first) + require.Len(t, second, 1, "copies split across subscriptions must still reach t+1 once") +} + +func TestOnSignResultLogModeDeliversUnsignedResults(t *testing.T) { + _, publicKeys := newTestNodes(t, 3) + queue, delivered := newSigningResultClient(t, ResultVerifyLog, publicKeys, 2) + body, err := json.Marshal(event.SigningResultEvent{ResultType: event.ResultTypeSuccess, TxID: "tx-1"}) + require.NoError(t, err) + + require.NoError(t, queue.handler(messaging.Message{Subject: event.SigningResultSubject(""), Data: body})) + require.Len(t, *delivered, 1) +} + +func TestOnSignResultOffDeliversEveryResult(t *testing.T) { + queue, delivered := newSigningResultClient(t, "", nil, 0) + body, err := json.Marshal(event.SigningResultEvent{ResultType: event.ResultTypeSuccess, TxID: "tx-1"}) + require.NoError(t, err) + + require.NoError(t, queue.handler(messaging.Message{Subject: event.SigningResultSubject(""), Data: body})) + require.NoError(t, queue.handler(messaging.Message{Subject: event.SigningResultSubject(""), Data: body})) + require.Len(t, *delivered, 2) +} + +func TestOnResharingResultUsesNewCommitteeThreshold(t *testing.T) { + nodes, publicKeys := newTestNodes(t, 3) + verifier, err := newResultVerifier(ResultVerifyEnforce, publicKeys, 2) + require.NoError(t, err) + queue := &capturingQueue{} + c := &mpcClient{reshareSuccessQueue: queue, resultVerifier: verifier, signer: staticSigner{}, pubsub: &recordingPubSub{}} + var delivered []event.ResharingResultEvent + require.NoError(t, c.OnResharingResult(func(e event.ResharingResultEvent) { delivered = append(delivered, e) })) + + require.NoError(t, c.Resharing(&types.ResharingMessage{SessionID: "session-1", NewThreshold: 2, WalletID: "wallet-1"})) + subject := event.ReshareResultSubject("", "session-1") + success := event.ResharingResultEvent{ResultType: event.ResultTypeSuccess, WalletID: "wallet-1", NewThreshold: 2, PubKey: []byte("pub")} + + require.NoError(t, queue.handler(signedMessage(t, nodes[0], subject, success))) + require.NoError(t, queue.handler(signedMessage(t, nodes[1], subject, success))) + require.Empty(t, delivered, "new threshold 2 needs 3 matching results") + require.NoError(t, queue.handler(signedMessage(t, nodes[2], subject, success))) + require.Len(t, delivered, 1) + + _, stillTracked := c.reshareRequiredSigs.Load(subject) + require.False(t, stillTracked) +} + +func TestNewResultVerifierValidation(t *testing.T) { + _, publicKeys := newTestNodes(t, 2) + tests := []struct { + name string + mode ResultVerifyMode + publicKeys map[string]string + required int + wantErr bool + }{ + {"off needs nothing", ResultVerifyOff, nil, 0, false}, + {"unknown mode", "strict", publicKeys, 1, true}, + {"enforce without keys", ResultVerifyEnforce, nil, 1, true}, + {"required above key count", ResultVerifyEnforce, publicKeys, 3, true}, + {"bad key", ResultVerifyLog, map[string]string{"a": "zz"}, 1, true}, + {"valid enforce", ResultVerifyEnforce, publicKeys, 2, false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + _, err := newResultVerifier(tt.mode, tt.publicKeys, tt.required) + require.Equal(t, tt.wantErr, err != nil, err) + }) + } +} + +type staticSigner struct{} + +func (staticSigner) Sign([]byte) ([]byte, error) { return []byte("sig"), nil } + +func (staticSigner) Algorithm() types.EventInitiatorKeyType { + return types.EventInitiatorKeyTypeEd25519 +} + +func (staticSigner) PublicKey() (string, error) { return "", nil } diff --git a/pkg/event/types.go b/pkg/event/types.go index 8bcb8116..efaf0fac 100644 --- a/pkg/event/types.go +++ b/pkg/event/types.go @@ -93,10 +93,11 @@ const ( ErrorCodeReshareFailure ErrorCode = "ERROR_RESHARE_FAILURE" // 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" + ErrorCodeContextCancelled ErrorCode = "ERROR_CONTEXT_CANCELLED" + ErrorCodeOperationAborted ErrorCode = "ERROR_OPERATION_ABORTED" + ErrorCodeNotMajority ErrorCode = "ERROR_NOT_MAJORITY" + ErrorCodeClusterNotReady ErrorCode = "ERROR_CLUSTER_NOT_READY" + ErrorCodeInsufficientNodeSignatures ErrorCode = "ERROR_INSUFFICIENT_NODE_SIGNATURES" ) // GetErrorCodeFromError attempts to categorize a generic error into a specific error code diff --git a/pkg/identity/identity.go b/pkg/identity/identity.go index 94241464..ba05b84a 100644 --- a/pkg/identity/identity.go +++ b/pkg/identity/identity.go @@ -460,6 +460,14 @@ func (s *fileStore) GetPublicKey(nodeID string) ([]byte, error) { return nil, fmt.Errorf("public key not found for node ID: %s", nodeID) } +// Sign signs arbitrary bytes with the node identity key; used for result signatures. +func (s *fileStore) Sign(data []byte) ([]byte, error) { + if len(s.privateKey) != ed25519.PrivateKeySize { + return nil, fmt.Errorf("node private key is not loaded") + } + return ed25519.Sign(s.privateKey, data), nil +} + func (s *fileStore) SignMessage(msg *types.TssMessage) ([]byte, error) { // Get deterministic bytes for signing msgBytes, err := msg.MarshalForSigning() diff --git a/pkg/messaging/message_queue.go b/pkg/messaging/message_queue.go index 445f4704..fe96847c 100644 --- a/pkg/messaging/message_queue.go +++ b/pkg/messaging/message_queue.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "sync" "time" "github.com/fystack/mpcium/pkg/logger" @@ -18,11 +19,21 @@ var ( type MessageQueue interface { Enqueue(topic string, message []byte, options *EnqueueOptions) error Dequeue(topic string, handler func(message []byte) error) error + DequeueMessage(topic string, handler func(msg Message) error) error + // DequeueMessageAsync settles each message only when done is called, with the same mapping as a handler error. + DequeueMessageAsync(topic string, handler func(msg Message, done func(err error))) error Close() } type EnqueueOptions struct { IdempotententKey string + Headers map[string]string +} + +type Message struct { + Subject string + Data []byte + Header nats.Header } type msgQueue struct { @@ -104,7 +115,10 @@ func (m *NATsMessageQueueManager) NewMessageQueue(consumerName, filterSubject st func (mq *msgQueue) Enqueue(topic string, message []byte, options *EnqueueOptions) error { header := nats.Header{} if options != nil { - header.Add("Nats-Msg-Id", options.IdempotententKey) + if options.IdempotententKey != "" { + header.Add("Nats-Msg-Id", options.IdempotententKey) + } + applyHeaders(header, options.Headers) } logger.Info("Publishing message", "topic", topic, "consumerName", mq.consumerName) @@ -123,34 +137,51 @@ func (mq *msgQueue) Enqueue(topic string, message []byte, options *EnqueueOption } func (mq *msgQueue) Dequeue(topic string, handler func(message []byte) error) error { + return mq.DequeueMessage(topic, func(msg Message) error { + return handler(msg.Data) + }) +} + +func (mq *msgQueue) DequeueMessage(topic string, handler func(msg Message) error) error { + return mq.DequeueMessageAsync(topic, func(msg Message, done func(err error)) { + done(handler(msg)) + }) +} + +func (mq *msgQueue) DequeueMessageAsync(topic string, handler func(msg Message, done func(err error))) error { c, err := mq.consumer.Consume(func(msg jetstream.Msg) { meta, _ := msg.Metadata() logger.Debug("Received message", "meta", meta) - err := handler(msg.Data()) - if err != nil { - if errors.Is(err, ErrPermament) { - logger.Info("Permanent error on message", "meta", meta) - if err := msg.Term(); err != nil { - logger.Error("Failed to terminate message", err) - } - return - } + var once sync.Once + handler(Message{Subject: msg.Subject(), Data: msg.Data(), Header: msg.Headers()}, func(err error) { + once.Do(func() { settleMessage(msg, meta, err) }) + }) + }) + mq.consumerContext = c + return err +} - logger.Error("Error handling message: ", err) - if err := msg.Nak(); err != nil { - logger.Error("Failed to nak message", err) +func settleMessage(msg jetstream.Msg, meta *jetstream.MsgMetadata, err error) { + if err != nil { + if errors.Is(err, ErrPermament) { + logger.Info("Permanent error on message", "meta", meta) + if err := msg.Term(); err != nil { + logger.Error("Failed to terminate message", err) } return } - logger.Debug("Message Acknowledged", "meta", meta) - err = msg.Ack() - if err != nil { - logger.Error("Error acknowledging message: ", err) + logger.Error("Error handling message: ", err) + if err := msg.Nak(); err != nil { + logger.Error("Failed to nak message", err) } - }) - mq.consumerContext = c - return err + return + } + + logger.Debug("Message Acknowledged", "meta", meta) + if err := msg.Ack(); err != nil { + logger.Error("Error acknowledging message: ", err) + } } func (mq *msgQueue) Close() { diff --git a/pkg/resultauth/aggregator.go b/pkg/resultauth/aggregator.go new file mode 100644 index 00000000..48f5ac73 --- /dev/null +++ b/pkg/resultauth/aggregator.go @@ -0,0 +1,219 @@ +package resultauth + +import ( + "crypto/sha256" + "sync" + "time" + + "github.com/fystack/mpcium/pkg/logger" +) + +const ( + DefaultAggregateTimeout = 15 * time.Second + // DefaultRememberedOutcomes bounds how many finished keys are kept to recognise late copies. + DefaultRememberedOutcomes = 50_000 +) + +type AggregatorConfig[T any] struct { + Timeout time.Duration + RememberedOutcomes int + Deliver func(T) + // Insufficient builds the error delivered when no success body reached the required count before the timeout. + Insufficient func(largest T, got, required int) T + // Completed, when set, is called once per round when its result is finalized. + Completed func(key string) +} + +// Result is one verified node copy; Ack settles its message once the copy is delivered or dropped. +type Result[T any] struct { + Key string + Required int + NodeID string + Body []byte + Success bool + Event T + Ack func() +} + +type successGroup[T any] struct { + event T + nodes map[string]struct{} +} + +type round[T any] struct { + required int + nodes map[string]struct{} + successes map[[sha256.Size]byte]*successGroup[T] + + firstError T + hasError bool + errorCount int + + acks []func() + timer *time.Timer +} + +// outcome remembers which nodes a finished key heard from: a further copy from one of them is a new attempt. +type outcome struct { + succeeded bool + nodes map[string]struct{} +} + +// Aggregator delivers one result per round, once Required nodes agree or the timeout fires; results must be verified. +type Aggregator[T any] struct { + cfg AggregatorConfig[T] + mu sync.Mutex + deliverMu sync.Mutex + rounds map[string]*round[T] + outcomes map[string]*outcome + order []string +} + +func NewAggregator[T any](cfg AggregatorConfig[T]) *Aggregator[T] { + if cfg.Timeout <= 0 { + cfg.Timeout = DefaultAggregateTimeout + } + if cfg.RememberedOutcomes <= 0 { + cfg.RememberedOutcomes = DefaultRememberedOutcomes + } + return &Aggregator[T]{ + cfg: cfg, + rounds: make(map[string]*round[T]), + outcomes: make(map[string]*outcome), + } +} + +func (a *Aggregator[T]) Add(result Result[T]) { + a.mu.Lock() + r, ok := a.rounds[result.Key] + if !ok { + if done, finished := a.outcomes[result.Key]; finished { + if _, repeated := done.nodes[result.NodeID]; !repeated { + done.nodes[result.NodeID] = struct{}{} + a.mu.Unlock() + settle(result.Ack) + return + } + } + r = &round[T]{ + required: max(result.Required, 1), + nodes: make(map[string]struct{}), + successes: make(map[[sha256.Size]byte]*successGroup[T]), + } + if r.required > 1 { + key := result.Key + r.timer = time.AfterFunc(a.cfg.Timeout, func() { a.expire(key) }) + } + a.rounds[result.Key] = r + } + + r.acks = append(r.acks, result.Ack) + if _, dup := r.nodes[result.NodeID]; dup { + a.mu.Unlock() + return + } + r.nodes[result.NodeID] = struct{}{} + + if !result.Success { + // Error payloads legitimately differ per node, so only distinct reporters are counted. + r.errorCount++ + if !r.hasError { + r.firstError, r.hasError = result.Event, true + } + if r.errorCount < r.required { + a.mu.Unlock() + return + } + a.finish(result.Key, r, false) + a.mu.Unlock() + a.complete(r, r.firstError) + return + } + + bodyHash := sha256.Sum256(result.Body) + group, ok := r.successes[bodyHash] + if !ok { + if len(r.successes) > 0 { + logger.Warn("Nodes sent different results for the same operation", "key", result.Key, "nodeID", result.NodeID) + } + group = &successGroup[T]{event: result.Event, nodes: make(map[string]struct{})} + r.successes[bodyHash] = group + } + group.nodes[result.NodeID] = struct{}{} + if len(group.nodes) < r.required { + a.mu.Unlock() + return + } + a.finish(result.Key, r, true) + a.mu.Unlock() + a.complete(r, group.event) +} + +// finish must be called with a.mu held. +func (a *Aggregator[T]) finish(key string, r *round[T], succeeded bool) { + if r.timer != nil { + r.timer.Stop() + } + delete(a.rounds, key) + if a.cfg.Completed != nil { + a.cfg.Completed(key) + } + if _, known := a.outcomes[key]; !known { + a.order = append(a.order, key) + if len(a.order) > a.cfg.RememberedOutcomes { + delete(a.outcomes, a.order[0]) + a.order = a.order[1:] + } + } + a.outcomes[key] = &outcome{succeeded: succeeded, nodes: r.nodes} +} + +// complete delivers outside a.mu, one result at a time, and settles the round's messages only afterwards. +func (a *Aggregator[T]) complete(r *round[T], result T) { + a.deliverMu.Lock() + a.cfg.Deliver(result) + a.deliverMu.Unlock() + for _, ack := range r.acks { + settle(ack) + } +} + +func (a *Aggregator[T]) expire(key string) { + a.mu.Lock() + r, ok := a.rounds[key] + if !ok { + a.mu.Unlock() + return + } + previous := a.outcomes[key] + a.finish(key, r, false) + if previous != nil && previous.succeeded { + // A round that misses quorum after a delivered success must not report that success as failed. + a.outcomes[key] = previous + a.mu.Unlock() + logger.Warn("Dropping result round that missed quorum after a delivered success", "key", key) + for _, ack := range r.acks { + settle(ack) + } + return + } + a.mu.Unlock() + + var largest *successGroup[T] + for _, group := range r.successes { + if largest == nil || len(group.nodes) > len(largest.nodes) { + largest = group + } + } + if largest != nil { + a.complete(r, a.cfg.Insufficient(largest.event, len(largest.nodes), r.required)) + return + } + a.complete(r, r.firstError) +} + +func settle(ack func()) { + if ack != nil { + ack() + } +} diff --git a/pkg/resultauth/aggregator_test.go b/pkg/resultauth/aggregator_test.go new file mode 100644 index 00000000..92fb59a1 --- /dev/null +++ b/pkg/resultauth/aggregator_test.go @@ -0,0 +1,213 @@ +package resultauth + +import ( + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +type testEvent struct { + Value string + Success bool + Reason string +} + +type deliveries struct { + mu sync.Mutex + events []testEvent + done chan struct{} +} + +func newDeliveries() *deliveries { return &deliveries{done: make(chan struct{}, 16)} } + +func (d *deliveries) deliver(ev testEvent) { + d.mu.Lock() + d.events = append(d.events, ev) + d.mu.Unlock() + d.done <- struct{}{} +} + +func (d *deliveries) snapshot() []testEvent { + d.mu.Lock() + defer d.mu.Unlock() + return append([]testEvent(nil), d.events...) +} + +func newTestAggregator(d *deliveries, timeout time.Duration) *Aggregator[testEvent] { + return NewAggregator(AggregatorConfig[testEvent]{ + Timeout: timeout, + Deliver: d.deliver, + Insufficient: func(first testEvent, got, required int) testEvent { + return testEvent{Value: first.Value, Success: false, Reason: "insufficient"} + }, + }) +} + +func successResult(node, body string) Result[testEvent] { + return Result[testEvent]{Key: "op-1", Required: 2, NodeID: node, Body: []byte(body), Success: true, Event: testEvent{Value: body, Success: true}} +} + +func errorResult(node, reason string) Result[testEvent] { + return Result[testEvent]{Key: "op-1", Required: 2, NodeID: node, Body: []byte(reason), Event: testEvent{Reason: reason}} +} + +func TestAggregatorDeliversAfterRequiredMatchingSuccesses(t *testing.T) { + d := newDeliveries() + a := newTestAggregator(d, time.Minute) + + a.Add(successResult("node0", "sig")) + require.Empty(t, d.snapshot()) + a.Add(successResult("node0", "sig")) + require.Empty(t, d.snapshot(), "the same node must not count twice") + a.Add(successResult("node1", "sig")) + a.Add(successResult("node2", "sig")) + + require.Equal(t, []testEvent{{Value: "sig", Success: true}}, d.snapshot()) +} + +func TestAggregatorIgnoresConflictingSuccess(t *testing.T) { + d := newDeliveries() + a := newTestAggregator(d, time.Minute) + + a.Add(successResult("node0", "sig")) + a.Add(successResult("node1", "forged")) + require.Empty(t, d.snapshot()) + a.Add(successResult("node2", "sig")) + + require.Equal(t, []testEvent{{Value: "sig", Success: true}}, d.snapshot()) +} + +func TestAggregatorForgedFirstCopyDoesNotBlockHonestQuorum(t *testing.T) { + d := newDeliveries() + a := newTestAggregator(d, time.Minute) + + a.Add(successResult("node0", "forged")) + a.Add(successResult("node1", "sig")) + a.Add(successResult("node2", "sig")) + + require.Equal(t, []testEvent{{Value: "sig", Success: true}}, d.snapshot()) +} + +func TestAggregatorSettlesCopiesAfterDelivery(t *testing.T) { + d := newDeliveries() + a := newTestAggregator(d, time.Minute) + var settled []string + withAck := func(r Result[testEvent]) Result[testEvent] { + r.Ack = func() { settled = append(settled, r.NodeID) } + return r + } + + a.Add(withAck(successResult("node0", "sig"))) + require.Empty(t, settled) + a.Add(withAck(successResult("node1", "sig"))) + require.Equal(t, []string{"node0", "node1"}, settled) + a.Add(withAck(successResult("node2", "sig"))) + require.Equal(t, []string{"node0", "node1", "node2"}, settled) +} + +func TestAggregatorLateCopyAfterTimeoutIsNotReportedAsFailure(t *testing.T) { + d := newDeliveries() + a := newTestAggregator(d, 20*time.Millisecond) + + a.Add(successResult("node0", "sig")) + a.Add(successResult("node1", "sig")) + time.Sleep(50 * time.Millisecond) + a.Add(successResult("node2", "sig")) + a.Add(successResult("node0", "forged-after-delivery")) + time.Sleep(50 * time.Millisecond) + + require.Equal(t, []testEvent{{Value: "sig", Success: true}}, d.snapshot()) +} + +func TestAggregatorRetryAfterErrorDelivers(t *testing.T) { + d := newDeliveries() + a := newTestAggregator(d, time.Minute) + + a.Add(errorResult("node0", "timeout")) + a.Add(errorResult("node1", "timeout")) + a.Add(successResult("node0", "sig")) + a.Add(successResult("node1", "sig")) + + require.Equal(t, []testEvent{{Reason: "timeout"}, {Value: "sig", Success: true}}, d.snapshot()) +} + +func TestAggregatorReplayFromSameNodesDeliversAgain(t *testing.T) { + d := newDeliveries() + a := newTestAggregator(d, time.Minute) + + a.Add(successResult("node0", "stored")) + a.Add(successResult("node1", "stored")) + a.Add(successResult("node0", "stored")) + a.Add(successResult("node1", "stored")) + + require.Len(t, d.snapshot(), 2) +} + +func TestAggregatorForgetsOldestOutcomes(t *testing.T) { + d := newDeliveries() + a := NewAggregator(AggregatorConfig[testEvent]{RememberedOutcomes: 2, Deliver: d.deliver, Insufficient: func(first testEvent, _, _ int) testEvent { return first }}) + for _, key := range []string{"op-1", "op-2", "op-3"} { + a.Add(Result[testEvent]{Key: key, Required: 1, NodeID: "node0", Body: []byte(key), Success: true}) + } + require.Len(t, a.outcomes, 2) + require.NotContains(t, a.outcomes, "op-1") +} + +func TestAggregatorDeliversErrorFromRequiredNodes(t *testing.T) { + d := newDeliveries() + a := newTestAggregator(d, time.Minute) + + a.Add(errorResult("node0", "timeout")) + a.Add(errorResult("node1", "aborted")) + + require.Equal(t, []testEvent{{Reason: "timeout"}}, d.snapshot()) +} + +func TestAggregatorTimesOutWithInsufficientSignatures(t *testing.T) { + d := newDeliveries() + a := newTestAggregator(d, 20*time.Millisecond) + + a.Add(successResult("node0", "sig")) + select { + case <-d.done: + case <-time.After(time.Second): + t.Fatal("aggregator did not time out") + } + + require.Equal(t, []testEvent{{Value: "sig", Reason: "insufficient"}}, d.snapshot()) +} + +func TestAggregatorTimesOutWithSingleError(t *testing.T) { + d := newDeliveries() + a := newTestAggregator(d, 20*time.Millisecond) + + a.Add(errorResult("node0", "timeout")) + select { + case <-d.done: + case <-time.After(time.Second): + t.Fatal("aggregator did not time out") + } + + require.Equal(t, []testEvent{{Reason: "timeout"}}, d.snapshot()) +} + +func TestAggregatorDropsLateResultsAndReportsCompletion(t *testing.T) { + d := newDeliveries() + var completed []string + a := NewAggregator(AggregatorConfig[testEvent]{ + Timeout: time.Minute, + Deliver: d.deliver, + Insufficient: func(first testEvent, _, _ int) testEvent { return first }, + Completed: func(key string) { completed = append(completed, key) }, + }) + + a.Add(successResult("node0", "sig")) + a.Add(successResult("node1", "sig")) + a.Add(successResult("node2", "sig")) + a.Add(errorResult("node2", "late")) + + require.Len(t, d.snapshot(), 1) + require.Equal(t, []string{"op-1"}, completed) +} diff --git a/pkg/resultauth/queue.go b/pkg/resultauth/queue.go new file mode 100644 index 00000000..568d7720 --- /dev/null +++ b/pkg/resultauth/queue.go @@ -0,0 +1,45 @@ +package resultauth + +import ( + "maps" + + "github.com/fystack/mpcium/pkg/messaging" +) + +type QueueOptions struct { + // PerNodeResults keeps every node's copy of a result so clients can count t+1 signatures. + PerNodeResults bool +} + +type signingQueue struct { + messaging.MessageQueue + signer Signer + nodeID string + opts QueueOptions +} + +// NewSigningQueue wraps a result queue so every enqueued result carries this node's signature headers. +func NewSigningQueue(queue messaging.MessageQueue, signer Signer, nodeID string, opts QueueOptions) messaging.MessageQueue { + return &signingQueue{MessageQueue: queue, signer: signer, nodeID: nodeID, opts: opts} +} + +func (q *signingQueue) Enqueue(topic string, message []byte, options *messaging.EnqueueOptions) error { + signatureHeaders, err := SignHeaders(q.signer, q.nodeID, topic, message) + if err != nil { + return err + } + + signedOptions := messaging.EnqueueOptions{} + if options != nil { + signedOptions = *options + } + if q.opts.PerNodeResults && signedOptions.IdempotententKey != "" { + signedOptions.IdempotententKey += ":" + q.nodeID + } + headers := make(map[string]string, len(signedOptions.Headers)+len(signatureHeaders)) + maps.Copy(headers, signedOptions.Headers) + maps.Copy(headers, signatureHeaders) + signedOptions.Headers = headers + + return q.MessageQueue.Enqueue(topic, message, &signedOptions) +} diff --git a/pkg/resultauth/resultauth.go b/pkg/resultauth/resultauth.go new file mode 100644 index 00000000..f7a58deb --- /dev/null +++ b/pkg/resultauth/resultauth.go @@ -0,0 +1,134 @@ +package resultauth + +import ( + "crypto/ed25519" + "encoding/base64" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + "strings" + + "github.com/fystack/mpcium/pkg/identity" +) + +const ( + NodeIDHeader = "Mpc-Node-Id" + SignatureHeader = "Mpc-Result-Sig" + + signingDomain = "mpcium-result-v1" +) + +var ( + ErrUnsigned = errors.New("result is not signed by a node") + ErrUnknownNode = errors.New("result is signed by an unknown node") + ErrInvalidSignature = errors.New("result signature is invalid") +) + +type Signer interface { + Sign(data []byte) ([]byte, error) +} + +type KeyLookup func(nodeID string) (ed25519.PublicKey, bool) + +// SigningPayload binds the signature to the subject so a result cannot be replayed on another topic. +func SigningPayload(subject string, body []byte) []byte { + payload := make([]byte, 0, len(signingDomain)+len(subject)+len(body)+2) + payload = append(payload, signingDomain...) + payload = append(payload, 0) + payload = append(payload, subject...) + payload = append(payload, 0) + return append(payload, body...) +} + +func SignHeaders(signer Signer, nodeID, subject string, body []byte) (map[string]string, error) { + if signer == nil || nodeID == "" { + return nil, errors.New("result signer and node ID are required") + } + signature, err := signer.Sign(SigningPayload(subject, body)) + if err != nil { + return nil, fmt.Errorf("failed to sign result: %w", err) + } + return map[string]string{ + NodeIDHeader: nodeID, + SignatureHeader: base64.StdEncoding.EncodeToString(signature), + }, nil +} + +// Verify checks the received bytes against the pinned key of the node named in the headers and returns that node ID. +func Verify(lookup KeyLookup, subject string, body []byte, header func(key string) string) (string, error) { + nodeID := header(NodeIDHeader) + encodedSignature := header(SignatureHeader) + if nodeID == "" || encodedSignature == "" { + return "", ErrUnsigned + } + publicKey, ok := lookup(nodeID) + if !ok { + return nodeID, fmt.Errorf("%w: %s", ErrUnknownNode, nodeID) + } + signature, err := base64.StdEncoding.DecodeString(encodedSignature) + if err != nil { + return nodeID, fmt.Errorf("%w: %v", ErrInvalidSignature, err) + } + if !ed25519.Verify(publicKey, SigningPayload(subject, body), signature) { + return nodeID, ErrInvalidSignature + } + return nodeID, nil +} + +func NewKeyLookup(publicKeys map[string]string) (KeyLookup, error) { + keys := make(map[string]ed25519.PublicKey, len(publicKeys)) + for nodeID, encodedKey := range publicKeys { + publicKey, err := hex.DecodeString(strings.TrimSpace(encodedKey)) + if err != nil { + return nil, fmt.Errorf("invalid public key for node %s: %w", nodeID, err) + } + if len(publicKey) != ed25519.PublicKeySize { + return nil, fmt.Errorf("invalid public key size for node %s: %d", nodeID, len(publicKey)) + } + keys[nodeID] = publicKey + } + return func(nodeID string) (ed25519.PublicKey, bool) { + publicKey, ok := keys[nodeID] + return publicKey, ok + }, nil +} + +// LoadPublicKeys reads node_id -> public_key from _identity.json files, for local setups and tests. +func LoadPublicKeys(identityDir string) (map[string]string, error) { + paths, err := filepath.Glob(filepath.Join(identityDir, "*_identity.json")) + if err != nil { + return nil, fmt.Errorf("failed to list identity files: %w", err) + } + publicKeys := make(map[string]string, len(paths)) + for _, path := range paths { + data, err := os.ReadFile(path) + if err != nil { + return nil, fmt.Errorf("failed to read %s: %w", path, err) + } + var nodeIdentity identity.NodeIdentity + if err := json.Unmarshal(data, &nodeIdentity); err != nil { + return nil, fmt.Errorf("failed to parse %s: %w", path, err) + } + if nodeIdentity.NodeID == "" || nodeIdentity.PublicKey == "" { + continue + } + publicKeys[nodeIdentity.NodeID] = nodeIdentity.PublicKey + } + return publicKeys, nil +} + +// LoadPublicKeysFile reads a node_id -> hex public key JSON map, as written by setup_identities.sh. +func LoadPublicKeysFile(path string) (map[string]string, error) { + data, err := os.ReadFile(path) + if err != nil { + return nil, err + } + var publicKeys map[string]string + if err := json.Unmarshal(data, &publicKeys); err != nil { + return nil, fmt.Errorf("failed to parse %s: %w", path, err) + } + return publicKeys, nil +} diff --git a/pkg/resultauth/resultauth_test.go b/pkg/resultauth/resultauth_test.go new file mode 100644 index 00000000..22026dd4 --- /dev/null +++ b/pkg/resultauth/resultauth_test.go @@ -0,0 +1,212 @@ +package resultauth + +import ( + "crypto/ed25519" + "crypto/rand" + "encoding/hex" + "encoding/json" + "os" + "path/filepath" + "testing" + + "github.com/fystack/mpcium/pkg/messaging" + "github.com/nats-io/nats.go" + "github.com/stretchr/testify/require" +) + +type keySigner struct{ privateKey ed25519.PrivateKey } + +func (s keySigner) Sign(data []byte) ([]byte, error) { return ed25519.Sign(s.privateKey, data), nil } + +func newTestNode(t *testing.T) (keySigner, string) { + t.Helper() + publicKey, privateKey, err := ed25519.GenerateKey(rand.Reader) + require.NoError(t, err) + return keySigner{privateKey: privateKey}, hex.EncodeToString(publicKey) +} + +func headerOf(headers map[string]string) func(string) string { + return func(key string) string { return headers[key] } +} + +func TestSignAndVerify(t *testing.T) { + signer, publicKey := newTestNode(t) + lookup, err := NewKeyLookup(map[string]string{"node0": publicKey}) + require.NoError(t, err) + + subject := "mpc.mpc_keygen_result.wallet-1" + body := []byte(`{"wallet_id":"wallet-1"}`) + headers, err := SignHeaders(signer, "node0", subject, body) + require.NoError(t, err) + + nodeID, err := Verify(lookup, subject, body, headerOf(headers)) + require.NoError(t, err) + require.Equal(t, "node0", nodeID) + + tests := []struct { + name string + subject string + body []byte + headers map[string]string + wantErr error + }{ + {"tampered body", subject, []byte(`{"wallet_id":"wallet-2"}`), headers, ErrInvalidSignature}, + {"replayed on another subject", "mpc.mpc_keygen_result.wallet-2", body, headers, ErrInvalidSignature}, + {"unsigned", subject, body, map[string]string{}, ErrUnsigned}, + {"unknown node", subject, body, map[string]string{NodeIDHeader: "node9", SignatureHeader: headers[SignatureHeader]}, ErrUnknownNode}, + {"malformed signature", subject, body, map[string]string{NodeIDHeader: "node0", SignatureHeader: "%%%"}, ErrInvalidSignature}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + _, err := Verify(lookup, tt.subject, tt.body, headerOf(tt.headers)) + require.ErrorIs(t, err, tt.wantErr) + }) + } +} + +func TestVerifyRejectsSignatureFromOtherNode(t *testing.T) { + signer0, publicKey0 := newTestNode(t) + _, publicKey1 := newTestNode(t) + lookup, err := NewKeyLookup(map[string]string{"node0": publicKey0, "node1": publicKey1}) + require.NoError(t, err) + + subject, body := "mpc.mpc_signing_result.complete", []byte(`{"tx_id":"tx-1"}`) + headers, err := SignHeaders(signer0, "node0", subject, body) + require.NoError(t, err) + headers[NodeIDHeader] = "node1" + + _, err = Verify(lookup, subject, body, headerOf(headers)) + require.ErrorIs(t, err, ErrInvalidSignature) +} + +func TestVectors(t *testing.T) { + data, err := os.ReadFile(filepath.Join("testdata", "vectors.json")) + require.NoError(t, err) + var vectors []struct { + Seed string `json:"seed"` + PublicKey string `json:"public_key"` + NodeID string `json:"node_id"` + Subject string `json:"subject"` + Body string `json:"body"` + Signature string `json:"signature"` + // Header names are part of the wire format: nats.Header.Get is case-sensitive. + NodeIDHeader string `json:"node_id_header"` + SignatureHeader string `json:"signature_header"` + } + require.NoError(t, json.Unmarshal(data, &vectors)) + require.NotEmpty(t, vectors) + + for _, v := range vectors { + require.Equal(t, v.NodeIDHeader, NodeIDHeader) + require.Equal(t, v.SignatureHeader, SignatureHeader) + seed, err := hex.DecodeString(v.Seed) + require.NoError(t, err) + privateKey := ed25519.NewKeyFromSeed(seed) + require.Equal(t, v.PublicKey, hex.EncodeToString(privateKey.Public().(ed25519.PublicKey))) + + headers, err := SignHeaders(keySigner{privateKey: privateKey}, v.NodeID, v.Subject, []byte(v.Body)) + require.NoError(t, err) + require.Equal(t, v.Signature, headers[SignatureHeader]) + + lookup, err := NewKeyLookup(map[string]string{v.NodeID: v.PublicKey}) + require.NoError(t, err) + _, err = Verify(lookup, v.Subject, []byte(v.Body), headerOf(map[string]string{ + NodeIDHeader: v.NodeID, SignatureHeader: v.Signature, + })) + require.NoError(t, err) + } +} + +func TestNewKeyLookupRejectsBadKeys(t *testing.T) { + _, err := NewKeyLookup(map[string]string{"node0": "not-hex"}) + require.Error(t, err) + _, err = NewKeyLookup(map[string]string{"node0": "abcd"}) + require.Error(t, err) +} + +func TestLoadPublicKeys(t *testing.T) { + dir := t.TempDir() + _, publicKey0 := newTestNode(t) + _, publicKey1 := newTestNode(t) + writeIdentity := func(name, nodeID, publicKey string) { + data, err := json.Marshal(map[string]string{"node_name": name, "node_id": nodeID, "public_key": publicKey}) + require.NoError(t, err) + require.NoError(t, os.WriteFile(filepath.Join(dir, name+"_identity.json"), data, 0o600)) + } + writeIdentity("node0", "id-0", publicKey0) + writeIdentity("node1", "id-1", publicKey1) + require.NoError(t, os.WriteFile(filepath.Join(dir, "node0_private.key"), []byte("secret"), 0o600)) + + publicKeys, err := LoadPublicKeys(dir) + require.NoError(t, err) + require.Equal(t, map[string]string{"id-0": publicKey0, "id-1": publicKey1}, publicKeys) +} + +type recordingQueue struct { + messaging.MessageQueue + topic string + message []byte + options *messaging.EnqueueOptions +} + +func (q *recordingQueue) Enqueue(topic string, message []byte, options *messaging.EnqueueOptions) error { + q.topic, q.message, q.options = topic, message, options + return nil +} + +func TestSigningQueue(t *testing.T) { + signer, publicKey := newTestNode(t) + lookup, err := NewKeyLookup(map[string]string{"node0": publicKey}) + require.NoError(t, err) + subject, body := "mpc.mpc_signing_result.complete", []byte(`{"tx_id":"tx-1"}`) + + tests := []struct { + name string + perNodeResults bool + options *messaging.EnqueueOptions + wantKey string + }{ + {"shared idempotency key", false, &messaging.EnqueueOptions{IdempotententKey: "sign:tx-1"}, "sign:tx-1"}, + {"per-node idempotency key", true, &messaging.EnqueueOptions{IdempotententKey: "sign:tx-1"}, "sign:tx-1:node0"}, + {"no options", true, nil, ""}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + inner := &recordingQueue{} + queue := NewSigningQueue(inner, signer, "node0", QueueOptions{PerNodeResults: tt.perNodeResults}) + require.NoError(t, queue.Enqueue(subject, body, tt.options)) + + require.Equal(t, tt.wantKey, inner.options.IdempotententKey) + header := nats.Header{} + for k, v := range inner.options.Headers { + header.Set(k, v) + } + nodeID, err := Verify(lookup, inner.topic, inner.message, header.Get) + require.NoError(t, err) + require.Equal(t, "node0", nodeID) + }) + } +} + +func TestSigningQueueKeepsCallerHeaders(t *testing.T) { + signer, _ := newTestNode(t) + inner := &recordingQueue{} + queue := NewSigningQueue(inner, signer, "node0", QueueOptions{}) + callerHeaders := map[string]string{"ClientID": "client-a"} + + require.NoError(t, queue.Enqueue("subject", []byte("body"), &messaging.EnqueueOptions{Headers: callerHeaders})) + require.Equal(t, "client-a", inner.options.Headers["ClientID"]) + require.Len(t, callerHeaders, 1) +} + +func TestLoadPublicKeysFile(t *testing.T) { + path := filepath.Join(t.TempDir(), "node_public_keys.json") + require.NoError(t, os.WriteFile(path, []byte(`{"id-0": "aa", "id-1": "bb"}`), 0o600)) + + publicKeys, err := LoadPublicKeysFile(path) + require.NoError(t, err) + require.Equal(t, map[string]string{"id-0": "aa", "id-1": "bb"}, publicKeys) + + _, err = LoadPublicKeysFile(filepath.Join(t.TempDir(), "missing.json")) + require.ErrorIs(t, err, os.ErrNotExist) +} diff --git a/pkg/resultauth/testdata/vectors.json b/pkg/resultauth/testdata/vectors.json new file mode 100644 index 00000000..dc3565e3 --- /dev/null +++ b/pkg/resultauth/testdata/vectors.json @@ -0,0 +1,12 @@ +[ + { + "seed": "000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f", + "public_key": "03a107bff3ce10be1d70dd18e74bc09967e4d6309ba50d5f1ddc8664125531b8", + "node_id": "node-vector-0", + "subject": "mpc.mpc_signing_result.complete", + "body": "{\"result_type\":\"success\",\"wallet_id\":\"wallet-1\",\"tx_id\":\"tx-1\"}", + "signature": "IJjFJSE7bykp+PEfdzZ9SGC1tcd2fht4JjLMX1obph6gWR7tgNxo1PG1bfeotiaMpK/+a4VDUzDp907XvJa5AA==", + "node_id_header": "Mpc-Node-Id", + "signature_header": "Mpc-Result-Sig" + } +] diff --git a/setup_identities.sh b/setup_identities.sh index 5bfbfe1f..960d30aa 100755 --- a/setup_identities.sh +++ b/setup_identities.sh @@ -109,6 +109,10 @@ for i in $(seq 0 $((NUM_NODES-1))); do done done +# Collect node public keys for clients that verify node-signed results +echo "🔑 Writing node_public_keys.json ..." +mpcium-cli export-node-keys --identity-dir node0/identity --peers peers.json --output node_public_keys.json --overwrite + echo "✨ Node identities setup complete!" echo echo "📂 Created folder structure:" From e04c57996dae3f9499b16db8db32af924610732d Mon Sep 17 00:00:00 2001 From: vietddude Date: Thu, 5 Nov 2026 15:37:34 +0700 Subject: [PATCH 2/2] Address code scanning findings on result signing - Build the signing payload with append instead of a summed capacity (CodeQL) - Write node_public_keys.json with 0600 (gosec) - Split Aggregator.Add into roundFor and record to cut its complexity (Sonar) --- cmd/mpcium-cli/export-node-keys.go | 2 +- pkg/resultauth/aggregator.go | 72 +++++++++++++++++------------- pkg/resultauth/resultauth.go | 4 +- 3 files changed, 44 insertions(+), 34 deletions(-) diff --git a/cmd/mpcium-cli/export-node-keys.go b/cmd/mpcium-cli/export-node-keys.go index 8c71765a..f49f83e4 100644 --- a/cmd/mpcium-cli/export-node-keys.go +++ b/cmd/mpcium-cli/export-node-keys.go @@ -50,7 +50,7 @@ func exportNodeKeys(ctx context.Context, c *cli.Command) error { if _, err := os.Stat(outputPath); err == nil && !c.Bool("overwrite") { return fmt.Errorf("file %s already exists, use --overwrite to replace it", outputPath) } - if err := os.WriteFile(outputPath, data, 0644); err != nil { + if err := os.WriteFile(outputPath, data, 0o600); err != nil { return fmt.Errorf("failed to write %s: %w", outputPath, err) } fmt.Printf("Wrote %d node public keys to %s\n", len(publicKeys), outputPath) diff --git a/pkg/resultauth/aggregator.go b/pkg/resultauth/aggregator.go index 48f5ac73..94a14062 100644 --- a/pkg/resultauth/aggregator.go +++ b/pkg/resultauth/aggregator.go @@ -85,35 +85,52 @@ func NewAggregator[T any](cfg AggregatorConfig[T]) *Aggregator[T] { func (a *Aggregator[T]) Add(result Result[T]) { a.mu.Lock() - r, ok := a.rounds[result.Key] - if !ok { - if done, finished := a.outcomes[result.Key]; finished { - if _, repeated := done.nodes[result.NodeID]; !repeated { - done.nodes[result.NodeID] = struct{}{} - a.mu.Unlock() - settle(result.Ack) - return - } - } - r = &round[T]{ - required: max(result.Required, 1), - nodes: make(map[string]struct{}), - successes: make(map[[sha256.Size]byte]*successGroup[T]), - } - if r.required > 1 { - key := result.Key - r.timer = time.AfterFunc(a.cfg.Timeout, func() { a.expire(key) }) - } - a.rounds[result.Key] = r + r := a.roundFor(result) + if r == nil { + a.mu.Unlock() + settle(result.Ack) + return } - r.acks = append(r.acks, result.Ack) if _, dup := r.nodes[result.NodeID]; dup { a.mu.Unlock() return } r.nodes[result.NodeID] = struct{}{} + finished, outcome := a.record(result, r) + a.mu.Unlock() + if finished { + a.complete(r, outcome) + } +} + +// roundFor returns the open round for the result, or nil for a late copy of a finished key. Must hold a.mu. +func (a *Aggregator[T]) roundFor(result Result[T]) *round[T] { + if r, ok := a.rounds[result.Key]; ok { + return r + } + if done, finished := a.outcomes[result.Key]; finished { + if _, repeated := done.nodes[result.NodeID]; !repeated { + done.nodes[result.NodeID] = struct{}{} + return nil + } + } + r := &round[T]{ + required: max(result.Required, 1), + nodes: make(map[string]struct{}), + successes: make(map[[sha256.Size]byte]*successGroup[T]), + } + if r.required > 1 { + key := result.Key + r.timer = time.AfterFunc(a.cfg.Timeout, func() { a.expire(key) }) + } + a.rounds[result.Key] = r + return r +} +// record counts one node's copy and finishes the round once it reaches quorum. Must hold a.mu. +func (a *Aggregator[T]) record(result Result[T], r *round[T]) (bool, T) { + var none T if !result.Success { // Error payloads legitimately differ per node, so only distinct reporters are counted. r.errorCount++ @@ -121,13 +138,10 @@ func (a *Aggregator[T]) Add(result Result[T]) { r.firstError, r.hasError = result.Event, true } if r.errorCount < r.required { - a.mu.Unlock() - return + return false, none } a.finish(result.Key, r, false) - a.mu.Unlock() - a.complete(r, r.firstError) - return + return true, r.firstError } bodyHash := sha256.Sum256(result.Body) @@ -141,12 +155,10 @@ func (a *Aggregator[T]) Add(result Result[T]) { } group.nodes[result.NodeID] = struct{}{} if len(group.nodes) < r.required { - a.mu.Unlock() - return + return false, none } a.finish(result.Key, r, true) - a.mu.Unlock() - a.complete(r, group.event) + return true, group.event } // finish must be called with a.mu held. diff --git a/pkg/resultauth/resultauth.go b/pkg/resultauth/resultauth.go index f7a58deb..3b240929 100644 --- a/pkg/resultauth/resultauth.go +++ b/pkg/resultauth/resultauth.go @@ -35,9 +35,7 @@ type KeyLookup func(nodeID string) (ed25519.PublicKey, bool) // SigningPayload binds the signature to the subject so a result cannot be replayed on another topic. func SigningPayload(subject string, body []byte) []byte { - payload := make([]byte, 0, len(signingDomain)+len(subject)+len(body)+2) - payload = append(payload, signingDomain...) - payload = append(payload, 0) + payload := append([]byte(signingDomain), 0) payload = append(payload, subject...) payload = append(payload, 0) return append(payload, body...)