Skip to content

Commit c9ba2f9

Browse files
martintomazicmartintomazic
authored
Merge pull request #6534 from oasisprotocol/martin/bugfix/penalize-faulty-checkpoints-peer
go/worker/storage/p2p: Penalize bad checkpoint sync peers
2 parents c3eed8d + 1d72eff commit c9ba2f9

9 files changed

Lines changed: 231 additions & 65 deletions

File tree

‎.changelog/6534.bugfix.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
go/worker/storage/p2p: Penalize bad checkpoint sync peers

‎go/consensus/cometbft/abci/snapshots.go‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -79,6 +79,12 @@ func (mux *abciMux) OfferSnapshot(req types.RequestOfferSnapshot) types.Response
7979
)
8080
return types.ResponseOfferSnapshot{Result: types.ResponseOfferSnapshot_REJECT}
8181
}
82+
if err := cp.Validate(); err != nil {
83+
mux.logger.Warn("received snapshot with invalid metadata",
84+
"err", err,
85+
)
86+
return types.ResponseOfferSnapshot{Result: types.ResponseOfferSnapshot_REJECT}
87+
}
8288

8389
// Number of chunks must match.
8490
if int(req.Snapshot.Chunks) != len(cp.Chunks) {

‎go/storage/mkvs/checkpoint/checkpoint.go‎

Lines changed: 0 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,6 @@ import (
66
"io"
77

88
"github.com/oasisprotocol/oasis-core/go/common"
9-
"github.com/oasisprotocol/oasis-core/go/common/crypto/hash"
109
"github.com/oasisprotocol/oasis-core/go/common/errors"
1110
"github.com/oasisprotocol/oasis-core/go/storage/mkvs/node"
1211
)
@@ -123,37 +122,3 @@ func NewCreateRestorer(creator Creator, restorer Restorer) CreateRestorer {
123122
Restorer: restorer,
124123
}
125124
}
126-
127-
// ChunkMetadata is chunk metadata.
128-
type ChunkMetadata struct {
129-
Version uint16 `json:"version"`
130-
Root node.Root `json:"root"`
131-
Index uint64 `json:"index"`
132-
Digest hash.Hash `json:"digest"`
133-
}
134-
135-
// Metadata is checkpoint metadata.
136-
type Metadata struct {
137-
Version uint16 `json:"version"`
138-
Root node.Root `json:"root"`
139-
Chunks []hash.Hash `json:"chunks"`
140-
}
141-
142-
// EncodedHash returns the encoded cryptographic hash of the checkpoint metadata.
143-
func (m *Metadata) EncodedHash() hash.Hash {
144-
return hash.NewFrom(m)
145-
}
146-
147-
// GetChunkMetadata returns the chunk metadata for the corresponding chunk.
148-
func (m Metadata) GetChunkMetadata(idx uint64) (*ChunkMetadata, error) {
149-
if idx >= uint64(len(m.Chunks)) {
150-
return nil, ErrChunkNotFound
151-
}
152-
153-
return &ChunkMetadata{
154-
Version: m.Version,
155-
Root: m.Root,
156-
Index: idx,
157-
Digest: m.Chunks[int(idx)],
158-
}, nil
159-
}

‎go/storage/mkvs/checkpoint/metadata.go‎

Lines changed: 56 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,56 @@
1+
package checkpoint
2+
3+
import (
4+
"fmt"
5+
6+
"github.com/oasisprotocol/oasis-core/go/common/crypto/hash"
7+
"github.com/oasisprotocol/oasis-core/go/storage/mkvs/node"
8+
)
9+
10+
// ChunkMetadata is chunk metadata.
11+
type ChunkMetadata struct {
12+
Version uint16 `json:"version"`
13+
Root node.Root `json:"root"`
14+
Index uint64 `json:"index"`
15+
Digest hash.Hash `json:"digest"`
16+
}
17+
18+
// Metadata is checkpoint metadata.
19+
type Metadata struct {
20+
Version uint16 `json:"version"`
21+
Root node.Root `json:"root"`
22+
Chunks []hash.Hash `json:"chunks"`
23+
}
24+
25+
// Validate checks that the metadata is structurally valid.
26+
func (m *Metadata) Validate() error {
27+
if m == nil {
28+
return fmt.Errorf("nil metadata")
29+
}
30+
if m.Root.Type == node.RootTypeInvalid || m.Root.Type > node.RootTypeMax {
31+
return fmt.Errorf("invalid root type: %s", m.Root.Type)
32+
}
33+
if len(m.Chunks) == 0 {
34+
return fmt.Errorf("zero chunks")
35+
}
36+
return nil
37+
}
38+
39+
// EncodedHash returns the encoded cryptographic hash of the checkpoint metadata.
40+
func (m *Metadata) EncodedHash() hash.Hash {
41+
return hash.NewFrom(m)
42+
}
43+
44+
// GetChunkMetadata returns the chunk metadata for the corresponding chunk.
45+
func (m Metadata) GetChunkMetadata(idx uint64) (*ChunkMetadata, error) {
46+
if idx >= uint64(len(m.Chunks)) {
47+
return nil, ErrChunkNotFound
48+
}
49+
50+
return &ChunkMetadata{
51+
Version: m.Version,
52+
Root: m.Root,
53+
Index: idx,
54+
Digest: m.Chunks[int(idx)],
55+
}, nil
56+
}

‎go/storage/mkvs/checkpoint/metadata_test.go‎

Lines changed: 59 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,59 @@
1+
package checkpoint
2+
3+
import (
4+
"testing"
5+
6+
"github.com/stretchr/testify/require"
7+
8+
"github.com/oasisprotocol/oasis-core/go/common/crypto/hash"
9+
"github.com/oasisprotocol/oasis-core/go/storage/mkvs"
10+
dbApi "github.com/oasisprotocol/oasis-core/go/storage/mkvs/db/api"
11+
"github.com/oasisprotocol/oasis-core/go/storage/mkvs/db/pathbadger"
12+
"github.com/oasisprotocol/oasis-core/go/storage/mkvs/node"
13+
)
14+
15+
func TestMetadataValidate(t *testing.T) {
16+
validRoot := node.Root{Namespace: testNs, Version: 1, Type: node.RootTypeIO}
17+
invalidRoot := node.Root{Namespace: testNs, Version: 1, Type: node.RootTypeInvalid}
18+
validChunks := []hash.Hash{{}}
19+
20+
tests := []struct {
21+
name string
22+
meta *Metadata
23+
wantErr string
24+
}{
25+
{name: "nil", meta: nil, wantErr: "nil"},
26+
{name: "zero chunks", meta: &Metadata{Root: validRoot}, wantErr: "zero chunks"},
27+
{name: "invalid root type", meta: &Metadata{Root: invalidRoot, Chunks: validChunks}, wantErr: "invalid root type"},
28+
}
29+
for _, test := range tests {
30+
t.Run(test.name, func(t *testing.T) {
31+
require.ErrorContains(t, test.meta.Validate(), test.wantErr)
32+
})
33+
}
34+
}
35+
36+
// TestMetadataValidateCpForEmptyState is a regression test asserting that a checkpoint created for
37+
// an empty root produces at least one chunk, so that Validate passes.
38+
func TestMetadataValidateCpForEmptyState(t *testing.T) {
39+
require := require.New(t)
40+
ctx := t.Context()
41+
42+
ndb, err := pathbadger.New(&dbApi.Config{Namespace: testNs, MemoryOnly: true})
43+
require.NoError(err, "New")
44+
defer ndb.Close()
45+
46+
tree := mkvs.New(nil, ndb, node.RootTypeIO)
47+
_, rootHash, err := tree.Commit(ctx, testNs, 1)
48+
require.NoError(err, "Commit")
49+
root := node.Root{Namespace: testNs, Version: 1, Type: node.RootTypeIO, Hash: rootHash}
50+
err = ndb.Finalize([]node.Root{root})
51+
require.NoError(err, "Finalize")
52+
53+
fc, err := NewFileCreator(t.TempDir(), ndb)
54+
require.NoError(err, "NewFileCreator")
55+
cp, err := fc.CreateCheckpoint(ctx, root, 16*1024, 0)
56+
require.NoError(err, "CreateCheckpoint")
57+
58+
require.NoError(cp.Validate())
59+
}

‎go/worker/storage/committee/checkpoint_sync.go‎

Lines changed: 46 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@ import (
1212
"time"
1313

1414
"github.com/oasisprotocol/oasis-core/go/p2p/rpc"
15+
"github.com/oasisprotocol/oasis-core/go/roothash/api/block"
1516
storageApi "github.com/oasisprotocol/oasis-core/go/storage/api"
1617
"github.com/oasisprotocol/oasis-core/go/storage/mkvs/checkpoint"
1718
"github.com/oasisprotocol/oasis-core/go/worker/storage/p2p/checkpointsync"
@@ -367,40 +368,30 @@ func sortCheckpoints(s []*checkpointsync.Checkpoint) {
367368
})
368369
}
369370

370-
func (w *Worker) checkCheckpointUsable(ctx context.Context, cp *checkpointsync.Checkpoint, remainingMask outstandingMask, genesisRound uint64) bool {
371-
namespace := w.commonNode.Runtime.ID()
372-
if !namespace.Equal(&cp.Root.Namespace) {
373-
// Not for the right runtime.
374-
return false
375-
}
376-
if cp.Root.Version == genesisRound && cp.Root.Type == storageApi.RootTypeIO {
377-
// Never fetch i/o root for genesis round.
378-
return false
371+
func validateCheckpoint(cp *checkpointsync.Checkpoint, blk *block.Block) error {
372+
if !blk.Header.Namespace.Equal((&cp.Root.Namespace)) {
373+
return fmt.Errorf("namespace mismatch: got %s, want %s", cp.Root.Namespace, blk.Header.Namespace)
379374
}
380375

381-
blk, err := w.commonNode.Runtime.History().GetCommittedBlock(ctx, cp.Root.Version)
382-
if err != nil {
383-
w.logger.Error("can't get block information for checkpoint, skipping", "err", err, "root", cp.Root)
384-
return false
376+
for _, root := range blk.Header.StorageRoots() {
377+
if cp.Root.Equal(&root) {
378+
return nil
379+
}
385380
}
381+
return fmt.Errorf("checkpoint metadata with unexpected root %s", cp.Root)
382+
}
383+
384+
func (w *Worker) checkCheckpointUsable(cp *checkpointsync.Checkpoint, remainingMask outstandingMask, genesisRound uint64) bool {
386385
_, lastIORoot, lastStateRoot := w.GetLastSynced()
387-
lastVersions := map[storageApi.RootType]uint64{
388-
storageApi.RootTypeIO: lastIORoot.Version,
389-
storageApi.RootTypeState: lastStateRoot.Version,
390-
}
391-
if namespace.Equal(&blk.Header.Namespace) {
392-
for _, root := range blk.Header.StorageRoots() {
393-
if cp.Root.Type == root.Type && root.Hash.Equal(&cp.Root.Hash) {
394-
// Do we already have this root?
395-
if lastVersions[cp.Root.Type] < cp.Root.Version && remainingMask.contains(cp.Root.Type) {
396-
return true
397-
}
398-
return false
399-
}
400-
}
386+
var lastVersion uint64
387+
switch cp.Root.Type {
388+
case storageApi.RootTypeIO:
389+
lastVersion = lastIORoot.Version
390+
case storageApi.RootTypeState:
391+
lastVersion = lastStateRoot.Version
401392
}
402-
w.logger.Info("checkpoint for unknown root skipped", "root", cp.Root)
403-
return false
393+
394+
return lastVersion < cp.Root.Version && remainingMask.contains(cp.Root.Type)
404395
}
405396

406397
func (w *Worker) syncCheckpoints(ctx context.Context, genesisRound uint64, wantOnlyGenesis bool) (*blockSummary, error) {
@@ -447,7 +438,32 @@ func (w *Worker) syncCheckpoints(ctx context.Context, genesisRound uint64, wantO
447438

448439
for _, check := range cps {
449440

450-
if check.Root.Version < genesisRound || !w.checkCheckpointUsable(ctx, check, remainingRoots, genesisRound) {
441+
if check.Root.Version < genesisRound {
442+
continue
443+
}
444+
445+
if check.Root.Version == genesisRound && check.Root.Type == storageApi.RootTypeIO {
446+
// Genesis round has no i/o root. Some peers may still advertise one after
447+
// dump-restore upgrades; ignore it without penalizing.
448+
continue
449+
}
450+
451+
blk, err := w.commonNode.Runtime.History().GetCommittedBlock(ctx, check.Root.Version)
452+
if err != nil {
453+
w.logger.Error("can't get block information for checkpoint, skipping", "err", err, "root", check.Root)
454+
continue
455+
}
456+
457+
if err := validateCheckpoint(check, blk); err != nil {
458+
w.logger.Error("invalid checkpoint received, penalizing bad peers", "checkpoint", check)
459+
for _, peer := range check.Peers {
460+
peer.RecordBadPeer()
461+
}
462+
continue
463+
}
464+
465+
if !w.checkCheckpointUsable(check, remainingRoots, genesisRound) {
466+
w.logger.Info("checkpoint not usable", "checkpoint", check)
451467
continue
452468
}
453469

‎go/worker/storage/committee/checkpoint_sync_test.go‎

Lines changed: 51 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,9 @@ import (
55

66
"github.com/stretchr/testify/assert"
77

8+
"github.com/oasisprotocol/oasis-core/go/common"
89
"github.com/oasisprotocol/oasis-core/go/p2p/rpc"
10+
"github.com/oasisprotocol/oasis-core/go/roothash/api/block"
911
"github.com/oasisprotocol/oasis-core/go/storage/mkvs/checkpoint"
1012
"github.com/oasisprotocol/oasis-core/go/storage/mkvs/node"
1113
"github.com/oasisprotocol/oasis-core/go/worker/storage/p2p/checkpointsync"
@@ -51,3 +53,52 @@ func TestSortCheckpoints(t *testing.T) {
5153

5254
assert.Equal(t, s, []*checkpointsync.Checkpoint{cp1, cp2, cp3, cp4})
5355
}
56+
57+
func TestValidateCheckpoint(t *testing.T) {
58+
runtimeID := common.NewTestNamespaceFromSeed([]byte("test namespace"), 0)
59+
blk := block.NewGenesisBlock(runtimeID, 0)
60+
61+
validRoot := blk.Header.StorageRootState()
62+
63+
wrongNamespaceRoot := validRoot
64+
wrongNamespaceRoot.Namespace = common.NewTestNamespaceFromSeed([]byte("test namespace invalid"), 0)
65+
66+
unexpectedRoot := validRoot
67+
unexpectedRoot.Hash[0] ^= 0xff // flip bits in the first byte so that hashes don't match.
68+
69+
for _, tc := range []struct {
70+
name string
71+
root node.Root
72+
errPrefix string
73+
}{
74+
{
75+
name: "valid root",
76+
root: validRoot,
77+
},
78+
{
79+
name: "namespace mismatch",
80+
root: wrongNamespaceRoot,
81+
errPrefix: "namespace mismatch:",
82+
},
83+
{
84+
name: "unexpected root",
85+
root: unexpectedRoot,
86+
errPrefix: "checkpoint metadata with unexpected root",
87+
},
88+
} {
89+
t.Run(tc.name, func(t *testing.T) {
90+
cp := &checkpointsync.Checkpoint{
91+
Metadata: &checkpoint.Metadata{
92+
Root: tc.root,
93+
},
94+
}
95+
96+
err := validateCheckpoint(cp, blk)
97+
if tc.errPrefix == "" {
98+
assert.NoError(t, err)
99+
return
100+
}
101+
assert.ErrorContains(t, err, tc.errPrefix)
102+
})
103+
}
104+
}

‎go/worker/storage/p2p/checkpointsync/client.go‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@ package checkpointsync
22

33
import (
44
"context"
5+
"slices"
56

67
"github.com/libp2p/go-libp2p/core"
78

@@ -59,6 +60,11 @@ func (c *client) GetCheckpoints(ctx context.Context, request *GetCheckpointsRequ
5960
for i, peerRsp := range rsps {
6061
peerCps := peerRsp.(*GetCheckpointsResponse).Checkpoints
6162

63+
if slices.ContainsFunc(peerCps, func(m *checkpoint.Metadata) bool { return m.Validate() != nil }) {
64+
pfs[i].RecordBadPeer()
65+
continue
66+
}
67+
6268
for _, cpMeta := range peerCps {
6369
h := cpMeta.EncodedHash()
6470
cp := cps[h]

‎go/worker/storage/p2p/synclegacy/client.go‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@ package synclegacy
22

33
import (
44
"context"
5+
"slices"
56

67
"github.com/libp2p/go-libp2p/core"
78

@@ -76,6 +77,11 @@ func (c *client) GetCheckpoints(ctx context.Context, request *GetCheckpointsRequ
7677
for i, peerRsp := range rsps {
7778
peerCps := peerRsp.(*GetCheckpointsResponse).Checkpoints
7879

80+
if slices.ContainsFunc(peerCps, func(m *checkpoint.Metadata) bool { return m.Validate() != nil }) {
81+
pfs[i].RecordBadPeer()
82+
continue
83+
}
84+
7985
for _, cpMeta := range peerCps {
8086
h := cpMeta.EncodedHash()
8187
cp := cps[h]

0 commit comments

Comments
 (0)