diff --git a/node/execution/ceremony/application/ceremony_application_validating.go b/node/execution/ceremony/application/ceremony_application_validating.go index d642983..4b50fac 100644 --- a/node/execution/ceremony/application/ceremony_application_validating.go +++ b/node/execution/ceremony/application/ceremony_application_validating.go @@ -4,6 +4,7 @@ import ( "bytes" "github.com/pkg/errors" + "golang.org/x/sync/errgroup" "source.quilibrium.com/quilibrium/monorepo/nekryptology/pkg/core/curves" "source.quilibrium.com/quilibrium/monorepo/node/protobufs" ) @@ -36,42 +37,59 @@ func (a *CeremonyApplication) applyTranscript( ) } - g1s := []*curves.PointBls48581G1{} + g1s := make([]*curves.PointBls48581G1, len(a.UpdatedTranscript.G1Powers)) + eg := errgroup.Group{} + eg.SetLimit(100) + for i := range a.UpdatedTranscript.G1Powers { - if !bytes.Equal( - a.UpdatedTranscript.G1Powers[i].KeyValue, - transcript.G1Powers[i].KeyValue, - ) { - return errors.Wrap(errors.New("invalid g1s"), "apply transcript") - } + i := i + eg.Go(func() error { + if !bytes.Equal( + a.UpdatedTranscript.G1Powers[i].KeyValue, + transcript.G1Powers[i].KeyValue, + ) { + return errors.Wrap(errors.New("invalid g1s"), "apply transcript") + } - g1 := &curves.PointBls48581G1{} - x, err := g1.FromAffineCompressed(a.UpdatedTranscript.G1Powers[i].KeyValue) - if err != nil { - return errors.Wrap(err, "apply transcript") - } - g1, _ = x.(*curves.PointBls48581G1) + g1 := &curves.PointBls48581G1{} + x, err := g1.FromAffineCompressed(a.UpdatedTranscript.G1Powers[i].KeyValue) + if err != nil { + return errors.Wrap(err, "apply transcript") + } + g1, _ = x.(*curves.PointBls48581G1) - g1s = append(g1s, g1) + g1s[i] = g1 + + return nil + }) } - g2s := []*curves.PointBls48581G2{} + g2s := make([]*curves.PointBls48581G2, len(a.UpdatedTranscript.G2Powers)) for i := range a.UpdatedTranscript.G2Powers { - if !bytes.Equal( - a.UpdatedTranscript.G2Powers[i].KeyValue, - transcript.G2Powers[i].KeyValue, - ) { - return errors.Wrap(errors.New("invalid g2s"), "apply transcript") - } + i := i + eg.Go(func() error { + if !bytes.Equal( + a.UpdatedTranscript.G2Powers[i].KeyValue, + transcript.G2Powers[i].KeyValue, + ) { + return errors.Wrap(errors.New("invalid g2s"), "apply transcript") + } - g2 := &curves.PointBls48581G2{} - x, err := g2.FromAffineCompressed(a.UpdatedTranscript.G2Powers[i].KeyValue) - if err != nil { - return errors.Wrap(err, "apply transcript") - } - g2, _ = x.(*curves.PointBls48581G2) + g2 := &curves.PointBls48581G2{} + x, err := g2.FromAffineCompressed(a.UpdatedTranscript.G2Powers[i].KeyValue) + if err != nil { + return errors.Wrap(err, "apply transcript") + } + g2, _ = x.(*curves.PointBls48581G2) - g2s = append(g2s, g2) + g2s[i] = g2 + + return nil + }) + } + + if err := eg.Wait(); err != nil { + return err } g1Witnesses := []*curves.PointBls48581G1{} diff --git a/node/execution/ceremony/application/oblivious_powers_of_tau.go b/node/execution/ceremony/application/oblivious_powers_of_tau.go index db843a7..f380748 100644 --- a/node/execution/ceremony/application/oblivious_powers_of_tau.go +++ b/node/execution/ceremony/application/oblivious_powers_of_tau.go @@ -70,7 +70,7 @@ func ProcessRound( for !participants[j].IsDone() { var err error if isReceiver { - msg, err = recv(seq, append(append([]byte{}, roundPeers[j]...), i...)) + msg, err = recv(seq, roundPeers[j]) if err != nil { return err } @@ -81,13 +81,13 @@ func ProcessRound( return err } - err = send(seq, append(append([]byte{}, i...), roundPeers[j]...), next) + err = send(seq, roundPeers[j], next) if err != nil { return err } if !isReceiver { - msg, err = recv(seq, append(append([]byte{}, roundPeers[j]...), i...)) + msg, err = recv(seq, roundPeers[j]) if err != nil { return err } diff --git a/node/execution/ceremony/application/oblivious_powers_of_tau_test.go b/node/execution/ceremony/application/oblivious_powers_of_tau_test.go index d097a2c..afbfdfc 100644 --- a/node/execution/ceremony/application/oblivious_powers_of_tau_test.go +++ b/node/execution/ceremony/application/oblivious_powers_of_tau_test.go @@ -196,36 +196,39 @@ func TestProcessRound(t *testing.T) { peerSecrets = append(peerSecrets, secrets) originalPeerSecrets = append(originalPeerSecrets, originalSecrets) - } messages := syncmap.Map{} - send := func(seq int, dst, msg []byte) error { - fmt.Printf("send %d bytes for seq %d to %+x\n", len(msg), seq, dst) + send := func(peer []byte) func(seq int, dst, msg []byte) error { + return func(seq int, dst, msg []byte) error { + fmt.Printf("send %d bytes for seq %d to %+x\n", len(msg), seq, dst) - b := byte(seq) - dst = append(append([]byte{}, b), dst...) - if msg == nil { - msg = []byte{0x01} + b := byte(seq) + dst = append(append(append([]byte{}, b), peer...), dst...) + if msg == nil { + msg = []byte{0x01} + } + messages.Store(string(dst), string(msg)) + return nil } - messages.Store(string(dst), string(msg)) - return nil } - recv := func(seq int, src []byte) ([]byte, error) { - fmt.Printf("recv %d from %+x\n", seq, src) + recv := func(peer []byte) func(seq int, src []byte) ([]byte, error) { + return func(seq int, src []byte) ([]byte, error) { + fmt.Printf("recv %d from %+x\n", seq, src) - b := byte(seq) - bsrc := append(append([]byte{}, b), src...) + b := byte(seq) + bsrc := append(append(append([]byte{}, b), src...), peer...) - msg, ok := messages.LoadAndDelete(string(bsrc)) - for !ok { - fmt.Printf("no message yet, waiting for recv %d from %+x\n", seq, src) + msg, ok := messages.LoadAndDelete(string(bsrc)) + for !ok { + fmt.Printf("no message yet, waiting for recv %d from %+x\n", seq, src) - time.Sleep(100 * time.Millisecond) - msg, ok = messages.LoadAndDelete(string(bsrc)) + time.Sleep(100 * time.Millisecond) + msg, ok = messages.LoadAndDelete(string(bsrc)) + } + + return []byte(msg.(string)), nil } - - return []byte(msg.(string)), nil } for j := 1; j < 4; j++ { @@ -244,8 +247,8 @@ func TestProcessRound(t *testing.T) { idkPoints, peerSecrets[i], curves.BLS48581G1(), - send, - recv, + send(peerPoints[i]), + recv(peerPoints[i]), []byte{0x01}, ) require.NoError(t, err) diff --git a/node/execution/ceremony/ceremony_execution_engine.go b/node/execution/ceremony/ceremony_execution_engine.go index cfafb9d..9dc6f95 100644 --- a/node/execution/ceremony/ceremony_execution_engine.go +++ b/node/execution/ceremony/ceremony_execution_engine.go @@ -45,6 +45,8 @@ type CeremonyExecutionEngine struct { activeClockFrame *protobufs.ClockFrame alreadyPublishedShare bool alreadyPublishedTranscript bool + seenMessageMap map[string]bool + seenMessageMx sync.Mutex } func NewCeremonyExecutionEngine( @@ -71,6 +73,8 @@ func NewCeremonyExecutionEngine( participantMx: sync.Mutex{}, peerChannels: map[string]*p2p.PublicP2PChannel{}, alreadyPublishedShare: false, + seenMessageMx: sync.Mutex{}, + seenMessageMap: map[string]bool{}, } provingKey, _, publicKeyBytes, provingKeyAddress := e.clock.GetProvingKey( @@ -329,7 +333,17 @@ func (e *CeremonyExecutionEngine) RunWorker() { zap.Uint64("current_round", app.RoundCount), ) - e.ensureSecrets(app) + if len(e.activeSecrets) == 0 { + // If we ended up in the scenario where we do not have any secrets + // available but we're in the round, we should politely leave. + for _, p := range app.ActiveParticipants { + if bytes.Equal(p.KeyValue, e.proverPublicKey) { + e.publishDroppedParticipant(e.proverPublicKey) + break + } + } + continue + } shouldConnect := false position := 0 @@ -339,6 +353,7 @@ func (e *CeremonyExecutionEngine) RunWorker() { if bytes.Equal(p.KeyValue, e.proverPublicKey) { shouldConnect = true position = i + break } } } @@ -378,6 +393,7 @@ func (e *CeremonyExecutionEngine) RunWorker() { err := e.participateRound(app) if err != nil { e.logger.Error("error while participating in round", zap.Error(err)) + e.publishDroppedParticipant(e.proverPublicKey) } } } else if len(app.ActiveParticipants) == 1 && @@ -411,7 +427,19 @@ func (e *CeremonyExecutionEngine) RunWorker() { } } - if !e.alreadyPublishedShare { + shouldPublish := false + for _, p := range app.ActiveParticipants { + if bytes.Equal(p.KeyValue, e.proverPublicKey) { + shouldPublish = true + break + } + } + + if !e.alreadyPublishedShare && shouldPublish { + if len(e.activeSecrets) == 0 { + e.publishDroppedParticipant(e.proverPublicKey) + continue + } err := e.publishTranscriptShare(app) if err != nil { e.logger.Error( @@ -620,8 +648,14 @@ func (e *CeremonyExecutionEngine) participateRound( return errors.Wrap(err, "participate round") } + spk, err := e.keyManager.GetAgreementKey("q-ratchet-spk") + if err != nil { + return errors.Wrap(err, "participate round") + } + idkPoint := curves.ED448().Point.Generator().Mul(idk) idks := []curves.Point{} + initiator := false for _, p := range app.ActiveParticipants { if !bytes.Equal(p.KeyValue, e.proverPublicKey) { ic, err := e.keyStore.GetLatestKeyBundle(p.KeyValue) @@ -647,8 +681,35 @@ func (e *CeremonyExecutionEngine) participateRound( if err != nil { return errors.Wrap(err, "participate round") } + + receiverSpk, err := curves.ED448().Point.FromAffineCompressed( + kba.SignedPreKey.GetPublicKeySignatureEd448().PublicKey.KeyValue, + ) + if err != nil { + return errors.Wrap(err, "participate round") + } + + if _, ok := e.peerChannels[string(p.KeyValue)]; !ok { + e.peerChannels[string(p.KeyValue)], err = p2p.NewPublicP2PChannel( + e.proverPublicKey, + p.KeyValue, + initiator, + idk, + spk, + receiverIdk, + receiverSpk, + curves.ED448(), + e.keyManager, + e.pubSub, + ) + if err != nil { + return errors.Wrap(err, "participate round") + } + } + idks = append(idks, receiverIdk) } else { + initiator = true idks = append(idks, idkPoint) } } @@ -666,19 +727,19 @@ func (e *CeremonyExecutionEngine) participateRound( idks, e.activeSecrets, curves.BLS48581G1(), - func(i int, filter, msg []byte) error { - return e.peerChannels[string(filter[len(e.proverPublicKey):])].Send(msg) + func(i int, receiver []byte, msg []byte) error { + return e.peerChannels[string(receiver)].Send(msg) }, - func(i int, filter []byte) ([]byte, error) { + func(i int, sender []byte) ([]byte, error) { msg, err := e.peerChannels[string( - filter[:len(e.proverPublicKey)], + sender, )].Receive() if err != nil { - e.publishDroppedParticipant(filter[:len(e.proverPublicKey)]) + e.publishDroppedParticipant(sender) return nil, err } else { if i == 0 { - e.publishLastSeenParticipant(filter[:len(e.proverPublicKey)]) + e.publishLastSeenParticipant(sender) } return msg, nil }