Skip to content

Commit 17f3d66

Browse files
joshua-kimjoshua-kim
committed
acp-118
Signed-off-by: Joshua Kim <20001595+joshua-kim@users.noreply.github.com>
1 parent e7648e5 commit 17f3d66

File tree

5 files changed

+583
-0
lines changed

5 files changed

+583
-0
lines changed

‎network/acp118/aggregator.go‎

Lines changed: 231 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,231 @@
1+
// Copyright (C) 2019-2024, Ava Labs, Inc. All rights reserved.
2+
// See the file LICENSE for licensing terms.
3+
4+
package acp118
5+
6+
import (
7+
"context"
8+
"errors"
9+
"fmt"
10+
"sync"
11+
12+
"go.uber.org/zap"
13+
"golang.org/x/sync/semaphore"
14+
"google.golang.org/protobuf/proto"
15+
16+
"github.com/ava-labs/avalanchego/codec"
17+
"github.com/ava-labs/avalanchego/ids"
18+
"github.com/ava-labs/avalanchego/network/p2p"
19+
"github.com/ava-labs/avalanchego/proto/pb/sdk"
20+
"github.com/ava-labs/avalanchego/utils/crypto/bls"
21+
"github.com/ava-labs/avalanchego/utils/logging"
22+
"github.com/ava-labs/avalanchego/utils/set"
23+
"github.com/ava-labs/avalanchego/vms/platformvm/warp"
24+
)
25+
26+
var (
27+
ErrDuplicateValidator = errors.New("duplicate validator")
28+
ErrInsufficientSignatures = errors.New("failed to aggregate sufficient stake weight of signatures")
29+
)
30+
31+
type result struct {
32+
message *warp.Message
33+
err error
34+
}
35+
36+
type Validator struct {
37+
NodeID ids.NodeID
38+
PublicKey *bls.PublicKey
39+
Weight uint64
40+
}
41+
42+
type indexedValidator struct {
43+
Validator
44+
I int
45+
}
46+
47+
// NewSignatureAggregator returns an instance of SignatureAggregator
48+
func NewSignatureAggregator(
49+
log logging.Logger,
50+
client *p2p.Client,
51+
maxPending int,
52+
) *SignatureAggregator {
53+
return &SignatureAggregator{
54+
log: log,
55+
client: client,
56+
maxPending: int64(maxPending),
57+
}
58+
}
59+
60+
// SignatureAggregator aggregates validator signatures for warp messages
61+
type SignatureAggregator struct {
62+
log logging.Logger
63+
client *p2p.Client
64+
codec codec.Codec
65+
maxPending int64
66+
}
67+
68+
// AggregateSignatures blocks until stakeWeightThreshold of validators signs the
69+
// provided message. Validators are issued requests in the caller-specified
70+
// order.
71+
func (s *SignatureAggregator) AggregateSignatures(
72+
parentCtx context.Context,
73+
message *warp.UnsignedMessage,
74+
justification []byte,
75+
validators []Validator,
76+
stakeWeightThreshold uint64,
77+
) (*warp.Message, error) {
78+
ctx, cancel := context.WithCancel(parentCtx)
79+
defer cancel()
80+
81+
request := &sdk.SignatureRequest{
82+
Message: message.Bytes(),
83+
Justification: justification,
84+
}
85+
86+
requestBytes, err := proto.Marshal(request)
87+
if err != nil {
88+
return nil, fmt.Errorf("failed to marshal signature request: %w", err)
89+
}
90+
91+
done := make(chan result)
92+
pendingRequests := semaphore.NewWeighted(s.maxPending)
93+
lock := &sync.Mutex{}
94+
aggregatedStakeWeight := uint64(0)
95+
attemptedStakeWeight := uint64(0)
96+
totalStakeWeight := uint64(0)
97+
signatures := make([]*bls.Signature, 0)
98+
signerBitSet := set.NewBits()
99+
100+
nodeIDsToValidator := make(map[ids.NodeID]indexedValidator)
101+
for i, v := range validators {
102+
totalStakeWeight += v.Weight
103+
104+
// Sanity check the validator set provided by the caller
105+
if _, ok := nodeIDsToValidator[v.NodeID]; ok {
106+
return nil, fmt.Errorf("%w: %s", ErrDuplicateValidator, v.NodeID)
107+
}
108+
109+
nodeIDsToValidator[v.NodeID] = indexedValidator{
110+
I: i,
111+
Validator: v,
112+
}
113+
}
114+
115+
onResponse := func(
116+
ctx context.Context,
117+
nodeID ids.NodeID,
118+
responseBytes []byte,
119+
err error,
120+
) {
121+
// We are guaranteed a response from a node in the validator set
122+
validator := nodeIDsToValidator[nodeID]
123+
124+
defer func() {
125+
lock.Lock()
126+
attemptedStakeWeight += validator.Weight
127+
remainingStakeWeight := totalStakeWeight - attemptedStakeWeight
128+
failed := remainingStakeWeight < stakeWeightThreshold
129+
lock.Unlock()
130+
131+
if failed {
132+
done <- result{err: ErrInsufficientSignatures}
133+
}
134+
135+
pendingRequests.Release(1)
136+
}()
137+
138+
if err != nil {
139+
s.log.Debug(
140+
"dropping response",
141+
zap.Stringer("nodeID", nodeID),
142+
zap.Error(err),
143+
)
144+
return
145+
}
146+
147+
response := &sdk.SignatureResponse{}
148+
if err := proto.Unmarshal(responseBytes, response); err != nil {
149+
s.log.Debug(
150+
"dropping response",
151+
zap.Stringer("nodeID", nodeID),
152+
zap.Error(err),
153+
)
154+
return
155+
}
156+
157+
signature, err := bls.SignatureFromBytes(response.Signature)
158+
if err != nil {
159+
s.log.Debug(
160+
"dropping response",
161+
zap.Stringer("nodeID", nodeID),
162+
zap.String("reason", "invalid signature"),
163+
zap.Error(err),
164+
)
165+
return
166+
}
167+
168+
if !bls.Verify(validator.PublicKey, signature, message.Bytes()) {
169+
s.log.Debug(
170+
"dropping response",
171+
zap.Stringer("nodeID", nodeID),
172+
zap.String("reason", "public key failed verification"),
173+
)
174+
return
175+
}
176+
177+
lock.Lock()
178+
signerBitSet.Add(validator.I)
179+
signatures = append(signatures, signature)
180+
aggregatedStakeWeight += validator.Weight
181+
182+
if aggregatedStakeWeight >= stakeWeightThreshold {
183+
aggregateSignature, err := bls.AggregateSignatures(signatures)
184+
if err != nil {
185+
done <- result{err: err}
186+
lock.Unlock()
187+
return
188+
}
189+
190+
bitSetSignature := &warp.BitSetSignature{
191+
Signers: signerBitSet.Bytes(),
192+
Signature: [bls.SignatureLen]byte{},
193+
}
194+
195+
copy(bitSetSignature.Signature[:], bls.SignatureToBytes(aggregateSignature))
196+
signedMessage, err := warp.NewMessage(message, bitSetSignature)
197+
done <- result{message: signedMessage, err: err}
198+
lock.Unlock()
199+
return
200+
}
201+
202+
lock.Unlock()
203+
}
204+
205+
for _, validator := range validators {
206+
if err := pendingRequests.Acquire(ctx, 1); err != nil {
207+
return nil, err
208+
}
209+
210+
// Avoid loop shadowing in goroutine
211+
validatorCopy := validator
212+
go func() {
213+
if err := s.client.AppRequest(
214+
ctx,
215+
set.Of(validatorCopy.NodeID),
216+
requestBytes,
217+
onResponse,
218+
); err != nil {
219+
done <- result{err: err}
220+
return
221+
}
222+
}()
223+
}
224+
225+
select {
226+
case <-ctx.Done():
227+
return nil, ctx.Err()
228+
case r := <-done:
229+
return r.message, r.err
230+
}
231+
}

‎network/acp118/aggregator_test.go‎

Lines changed: 123 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,123 @@
1+
// Copyright (C) 2019-2024, Ava Labs, Inc. All rights reserved.
2+
// See the file LICENSE for licensing terms.
3+
4+
package acp118
5+
6+
import (
7+
"context"
8+
"errors"
9+
"testing"
10+
11+
"github.com/stretchr/testify/require"
12+
13+
"github.com/ava-labs/avalanchego/ids"
14+
"github.com/ava-labs/avalanchego/network/p2p"
15+
"github.com/ava-labs/avalanchego/network/p2p/p2ptest"
16+
"github.com/ava-labs/avalanchego/utils/crypto/bls"
17+
"github.com/ava-labs/avalanchego/utils/logging"
18+
"github.com/ava-labs/avalanchego/vms/platformvm/warp"
19+
)
20+
21+
func TestVerifier_Verify(t *testing.T) {
22+
nodeID0 := ids.GenerateTestNodeID()
23+
sk0, err := bls.NewSecretKey()
24+
require.NoError(t, err)
25+
pk0 := bls.PublicFromSecretKey(sk0)
26+
networkID := uint32(123)
27+
chainID := ids.GenerateTestID()
28+
signer := warp.NewSigner(sk0, networkID, chainID)
29+
30+
tests := []struct {
31+
name string
32+
33+
ctx context.Context
34+
validators []Validator
35+
threshold uint64
36+
37+
handler p2p.Handler
38+
39+
wantErr error
40+
}{
41+
{
42+
name: "passes verification",
43+
ctx: context.Background(),
44+
validators: []Validator{
45+
{
46+
NodeID: nodeID0,
47+
PublicKey: pk0,
48+
Weight: 1,
49+
},
50+
},
51+
threshold: 1,
52+
handler: NewHandler(&testAttestor{}, signer, networkID, chainID),
53+
},
54+
{
55+
name: "fails verification",
56+
ctx: context.Background(),
57+
validators: []Validator{
58+
{
59+
NodeID: nodeID0,
60+
PublicKey: pk0,
61+
Weight: 1,
62+
},
63+
},
64+
threshold: 1,
65+
handler: NewHandler(
66+
&testAttestor{Err: errors.New("foobar")},
67+
signer,
68+
networkID,
69+
chainID,
70+
),
71+
wantErr: ErrInsufficientSignatures,
72+
},
73+
{
74+
name: "invalid validator set",
75+
ctx: context.Background(),
76+
validators: []Validator{
77+
{
78+
NodeID: nodeID0,
79+
PublicKey: pk0,
80+
Weight: 1,
81+
},
82+
{
83+
NodeID: nodeID0,
84+
PublicKey: pk0,
85+
Weight: 1,
86+
},
87+
},
88+
wantErr: ErrDuplicateValidator,
89+
},
90+
{
91+
name: "context canceled",
92+
ctx: func() context.Context {
93+
ctx, cancel := context.WithCancel(context.Background())
94+
cancel()
95+
96+
return ctx
97+
}(),
98+
threshold: 1,
99+
wantErr: context.Canceled,
100+
},
101+
}
102+
103+
for _, tt := range tests {
104+
t.Run(tt.name, func(t *testing.T) {
105+
require := require.New(t)
106+
107+
ctx := context.Background()
108+
message, err := warp.NewUnsignedMessage(networkID, chainID, []byte("payload"))
109+
require.NoError(err)
110+
client := p2ptest.NewClient(t, ctx, tt.handler, ids.GenerateTestNodeID(), nodeID0)
111+
verifier := NewSignatureAggregator(logging.NoLog{}, client, 1)
112+
113+
_, err = verifier.AggregateSignatures(
114+
tt.ctx,
115+
message,
116+
[]byte("justification"),
117+
tt.validators,
118+
tt.threshold,
119+
)
120+
require.ErrorIs(err, tt.wantErr)
121+
})
122+
}
123+
}

0 commit comments

Comments
 (0)