diff --git a/pkg/server/service/indexer/indexer.go b/pkg/server/service/indexer/indexer.go index 108e876..c64f9ba 100644 --- a/pkg/server/service/indexer/indexer.go +++ b/pkg/server/service/indexer/indexer.go @@ -177,6 +177,14 @@ func (i *Indexer) CreateBeaconState(ctx context.Context, req *indexer.CreateBeac return nil, status.Error(codes.Internal, err.Error()) } + // A location must never be shared between two different beacon states. Without + // this check, a request for an unrelated node/slot/state_root but a location + // that already belongs to another record would ride along on that record's + // blob, and later cause it to be deleted out from under the original record. + if err := i.checkBeaconStateLocationOwnership(ctx, req); err != nil { + return nil, err + } + if exists { // Check if the state is already indexed filter := &persistence.BeaconStateFilter{} @@ -435,6 +443,14 @@ func (i *Indexer) CreateBeaconBlock(ctx context.Context, req *indexer.CreateBeac return nil, status.Error(codes.Internal, err.Error()) } + // A location must never be shared between two different beacon blocks. Without + // this check, a request for an unrelated node/slot/block_root but a location + // that already belongs to another record would ride along on that record's + // blob, and later cause it to be deleted out from under the original record. + if err := i.checkBeaconBlockLocationOwnership(ctx, req); err != nil { + return nil, err + } + if exists { // Check if the block is already indexed filter := &persistence.BeaconBlockFilter{} @@ -701,6 +717,15 @@ func (i *Indexer) CreateBeaconBadBlock(ctx context.Context, req *indexer.CreateB return nil, status.Error(codes.Internal, err.Error()) } + // A location must never be shared between two different beacon bad blocks. + // Without this check, a request for an unrelated node/slot/block_root but a + // location that already belongs to another record would ride along on that + // record's blob, and later cause it to be deleted out from under the original + // record. + if err := i.checkBeaconBadBlockLocationOwnership(ctx, req); err != nil { + return nil, err + } + if exists { // Check if the bad block is already indexed filter := &persistence.BeaconBadBlockFilter{} @@ -959,6 +984,15 @@ func (i *Indexer) CreateBeaconBadBlob(ctx context.Context, req *indexer.CreateBe return nil, status.Error(codes.Internal, err.Error()) } + // A location must never be shared between two different beacon bad blobs. + // Without this check, a request for an unrelated node/slot/block_root/index but + // a location that already belongs to another record would ride along on that + // record's blob, and later cause it to be deleted out from under the original + // record. + if err := i.checkBeaconBadBlobLocationOwnership(ctx, req); err != nil { + return nil, err + } + if exists { // Check if the bad blob is already indexed filter := &persistence.BeaconBadBlobFilter{} @@ -1219,6 +1253,14 @@ func (i *Indexer) CreateExecutionBlockTrace(ctx context.Context, req *indexer.Cr return nil, status.Error(codes.InvalidArgument, err.Error()) } + // A location must never be shared between two different execution block traces. + // Without this check, a request for an unrelated node/block_hash but a location + // that already belongs to another record would ride along on that record's + // blob, and later cause it to be deleted out from under the original record. + if err := i.checkExecutionBlockTraceLocationOwnership(ctx, req); err != nil { + return nil, err + } + // Create the execution block trace trace := &indexer.ExecutionBlockTrace{ Id: wrapperspb.String(uuid.New().String()), @@ -1422,6 +1464,14 @@ func (i *Indexer) CreateExecutionBadBlock(ctx context.Context, req *indexer.Crea return nil, status.Error(codes.InvalidArgument, err.Error()) } + // A location must never be shared between two different execution bad blocks. + // Without this check, a request for an unrelated node/block_hash but a location + // that already belongs to another record would ride along on that record's + // blob, and later cause it to be deleted out from under the original record. + if err := i.checkExecutionBadBlockLocationOwnership(ctx, req); err != nil { + return nil, err + } + // Create the execution bad block block := &indexer.ExecutionBadBlock{ Id: wrapperspb.String(uuid.New().String()), diff --git a/pkg/server/service/indexer/location_ownership.go b/pkg/server/service/indexer/location_ownership.go new file mode 100644 index 0000000..34d1750 --- /dev/null +++ b/pkg/server/service/indexer/location_ownership.go @@ -0,0 +1,183 @@ +package indexer + +import ( + "context" + + "github.com/ethpandaops/tracoor/pkg/proto/tracoor/indexer" + "github.com/ethpandaops/tracoor/pkg/server/persistence" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +// These checks stop a create request from claiming a location that is +// already associated with a different record. A location is expected to be +// unique to the record that first uploaded a blob there; if two records with +// different identities are allowed to share one location, deleting either +// record (for example, once it ages out of retention) deletes the shared +// blob out from under the other one, even though that other record is still +// active and was never meant to be touched. + +func (i *Indexer) checkBeaconStateLocationOwnership(ctx context.Context, req *indexer.CreateBeaconStateRequest) error { + filter := &persistence.BeaconStateFilter{} + filter.AddLocation(req.GetLocation().GetValue()) + + existing, err := i.db.ListBeaconState(ctx, filter, &persistence.PaginationCursor{Limit: 1, Offset: 0}) + if err != nil { + return status.Error(codes.Internal, err.Error()) + } + + if len(existing) == 0 { + return nil + } + + record := existing[0] + + //nolint:gosec // slot is well within int64 range + sameRecord := record.Node == req.GetNode().GetValue() && + record.Network == req.GetNetwork().GetValue() && + record.Slot == int64(req.GetSlot().GetValue()) && + record.StateRoot == req.GetStateRoot().GetValue() + + if !sameRecord { + return status.Error(codes.AlreadyExists, "location is already associated with a different beacon state") + } + + return nil +} + +func (i *Indexer) checkBeaconBlockLocationOwnership(ctx context.Context, req *indexer.CreateBeaconBlockRequest) error { + filter := &persistence.BeaconBlockFilter{} + filter.AddLocation(req.GetLocation().GetValue()) + + existing, err := i.db.ListBeaconBlock(ctx, filter, &persistence.PaginationCursor{Limit: 1, Offset: 0}) + if err != nil { + return status.Error(codes.Internal, err.Error()) + } + + if len(existing) == 0 { + return nil + } + + record := existing[0] + + //nolint:gosec // slot is well within int64 range + sameRecord := record.Node == req.GetNode().GetValue() && + record.Network == req.GetNetwork().GetValue() && + record.Slot == int64(req.GetSlot().GetValue()) && + record.BlockRoot == req.GetBlockRoot().GetValue() + + if !sameRecord { + return status.Error(codes.AlreadyExists, "location is already associated with a different beacon block") + } + + return nil +} + +func (i *Indexer) checkBeaconBadBlockLocationOwnership(ctx context.Context, req *indexer.CreateBeaconBadBlockRequest) error { + filter := &persistence.BeaconBadBlockFilter{} + filter.AddLocation(req.GetLocation().GetValue()) + + existing, err := i.db.ListBeaconBadBlock(ctx, filter, &persistence.PaginationCursor{Limit: 1, Offset: 0}) + if err != nil { + return status.Error(codes.Internal, err.Error()) + } + + if len(existing) == 0 { + return nil + } + + record := existing[0] + + //nolint:gosec // slot is well within int64 range + sameRecord := record.Node == req.GetNode().GetValue() && + record.Network == req.GetNetwork().GetValue() && + record.Slot == int64(req.GetSlot().GetValue()) && + record.BlockRoot == req.GetBlockRoot().GetValue() + + if !sameRecord { + return status.Error(codes.AlreadyExists, "location is already associated with a different beacon bad block") + } + + return nil +} + +func (i *Indexer) checkBeaconBadBlobLocationOwnership(ctx context.Context, req *indexer.CreateBeaconBadBlobRequest) error { + filter := &persistence.BeaconBadBlobFilter{} + filter.AddLocation(req.GetLocation().GetValue()) + + existing, err := i.db.ListBeaconBadBlob(ctx, filter, &persistence.PaginationCursor{Limit: 1, Offset: 0}) + if err != nil { + return status.Error(codes.Internal, err.Error()) + } + + if len(existing) == 0 { + return nil + } + + record := existing[0] + + //nolint:gosec // slot and index are well within int64 range + sameRecord := record.Node == req.GetNode().GetValue() && + record.Network == req.GetNetwork().GetValue() && + record.Slot == int64(req.GetSlot().GetValue()) && + record.BlockRoot == req.GetBlockRoot().GetValue() && + record.Index == int64(req.GetIndex().GetValue()) + + if !sameRecord { + return status.Error(codes.AlreadyExists, "location is already associated with a different beacon bad blob") + } + + return nil +} + +func (i *Indexer) checkExecutionBlockTraceLocationOwnership(ctx context.Context, req *indexer.CreateExecutionBlockTraceRequest) error { + filter := &persistence.ExecutionBlockTraceFilter{} + filter.AddLocation(req.GetLocation().GetValue()) + + existing, err := i.db.ListExecutionBlockTrace(ctx, filter, &persistence.PaginationCursor{Limit: 1, Offset: 0}) + if err != nil { + return status.Error(codes.Internal, err.Error()) + } + + if len(existing) == 0 { + return nil + } + + record := existing[0] + + sameRecord := record.Node == req.GetNode().GetValue() && + record.Network == req.GetNetwork().GetValue() && + record.BlockHash == req.GetBlockHash().GetValue() + + if !sameRecord { + return status.Error(codes.AlreadyExists, "location is already associated with a different execution block trace") + } + + return nil +} + +func (i *Indexer) checkExecutionBadBlockLocationOwnership(ctx context.Context, req *indexer.CreateExecutionBadBlockRequest) error { + filter := &persistence.ExecutionBadBlockFilter{} + filter.AddLocation(req.GetLocation().GetValue()) + + existing, err := i.db.ListExecutionBadBlock(ctx, filter, &persistence.PaginationCursor{Limit: 1, Offset: 0}) + if err != nil { + return status.Error(codes.Internal, err.Error()) + } + + if len(existing) == 0 { + return nil + } + + record := existing[0] + + sameRecord := record.Node == req.GetNode().GetValue() && + record.Network == req.GetNetwork().GetValue() && + record.BlockHash == req.GetBlockHash().GetValue() + + if !sameRecord { + return status.Error(codes.AlreadyExists, "location is already associated with a different execution bad block") + } + + return nil +} diff --git a/pkg/server/service/indexer/location_ownership_test.go b/pkg/server/service/indexer/location_ownership_test.go new file mode 100644 index 0000000..85ac43f --- /dev/null +++ b/pkg/server/service/indexer/location_ownership_test.go @@ -0,0 +1,212 @@ +package indexer + +import ( + "context" + "fmt" + "os" + "testing" + "time" + + pindexer "github.com/ethpandaops/tracoor/pkg/proto/tracoor/indexer" + "github.com/ethpandaops/tracoor/pkg/server/ethereum" + "github.com/ethpandaops/tracoor/pkg/server/persistence" + "github.com/ethpandaops/tracoor/pkg/store" + "github.com/sirupsen/logrus" + "google.golang.org/protobuf/types/known/timestamppb" + "google.golang.org/protobuf/types/known/wrapperspb" +) + +// newLocationOwnershipTestIndexer builds a real Indexer backed by a +// file-backed SQLite database and an FS store, avoiding the Docker/Minio +// dependency that NewMockIndexer requires. +func newLocationOwnershipTestIndexer(t *testing.T) *Indexer { + t.Helper() + + ctx := context.Background() + + dbFile, err := os.CreateTemp("", "location_ownership_*.db") + if err != nil { + t.Fatalf("failed to create temp db file: %v", err) + } + dbPath := dbFile.Name() + dbFile.Close() + os.Remove(dbPath) + + t.Cleanup(func() { + os.Remove(dbPath) + os.Remove(dbPath + "-wal") + os.Remove(dbPath + "-shm") + }) + + db, err := persistence.NewIndexer("location-ownership-test", logrus.New(), persistence.Config{ + DSN: fmt.Sprintf("file:%s?parseTime=True", dbPath), + DriverName: "sqlite", + }, persistence.DefaultOptions().SetMetricsEnabled(false)) + if err != nil { + t.Fatalf("failed to create persistence indexer: %v", err) + } + if err := db.Start(ctx); err != nil { + t.Fatalf("failed to migrate: %v", err) + } + + basePath, err := os.MkdirTemp("", "location_ownership_fs") + if err != nil { + t.Fatalf("failed to create temp fs dir: %v", err) + } + t.Cleanup(func() { os.RemoveAll(basePath) }) + + st, err := store.NewFSStore("location-ownership-test", logrus.New(), &store.FSStoreConfig{BasePath: basePath}, &store.Options{}) + if err != nil { + t.Fatalf("failed to create FS store: %v", err) + } + + idx, err := NewIndexer(ctx, logrus.New(), &Config{}, db, st, ðereum.Config{}) + if err != nil { + t.Fatalf("failed to create indexer: %v", err) + } + + return idx +} + +func TestCreateBeaconState_RejectsLocationOwnedByADifferentRecord(t *testing.T) { + idx := newLocationOwnershipTestIndexer(t) + ctx := context.Background() + + location := "beacon_state/shared-location.ssz" + data := []byte("original owner's data") + + if _, err := idx.Store().SaveBeaconState(ctx, &store.SaveParams{Data: &data, Location: location}); err != nil { + t.Fatalf("failed to pre-upload blob: %v", err) + } + + // The legitimate record. + if _, err := idx.CreateBeaconState(ctx, &pindexer.CreateBeaconStateRequest{ + Node: wrapperspb.String("owner-node"), + Slot: wrapperspb.UInt64(100), + Epoch: wrapperspb.UInt64(3), + StateRoot: wrapperspb.String("0xowner"), + FetchedAt: timestamppb.New(time.Now()), + BeaconImplementation: wrapperspb.String("teku"), + NodeVersion: wrapperspb.String("1.0.0"), + Location: wrapperspb.String(location), + Network: wrapperspb.String("mainnet"), + }); err != nil { + t.Fatalf("legitimate create failed: %v", err) + } + + // A second record with a completely different identity, but the same + // location, must be rejected. + _, err := idx.CreateBeaconState(ctx, &pindexer.CreateBeaconStateRequest{ + Node: wrapperspb.String("someone-elses-node"), + Slot: wrapperspb.UInt64(1), + Epoch: wrapperspb.UInt64(0), + StateRoot: wrapperspb.String("0xdifferent"), + FetchedAt: timestamppb.New(time.Now()), + BeaconImplementation: wrapperspb.String("teku"), + NodeVersion: wrapperspb.String("1.0.0"), + Location: wrapperspb.String(location), + Network: wrapperspb.String("mainnet"), + }) + if err == nil { + t.Fatal("expected the colliding create to be rejected, got no error") + } + + countRsp, err := idx.CountBeaconState(ctx, &pindexer.CountBeaconStateRequest{}) + if err != nil { + t.Fatalf("failed to count: %v", err) + } + if countRsp.Count.Value != 1 { + t.Fatalf("expected exactly 1 row after the rejected collision attempt, got %d", countRsp.Count.Value) + } +} + +func TestCreateBeaconState_AllowsRetryOfTheSameRecord(t *testing.T) { + idx := newLocationOwnershipTestIndexer(t) + ctx := context.Background() + + location := "beacon_state/retry.ssz" + data := []byte("data") + + if _, err := idx.Store().SaveBeaconState(ctx, &store.SaveParams{Data: &data, Location: location}); err != nil { + t.Fatalf("failed to pre-upload blob: %v", err) + } + + req := &pindexer.CreateBeaconStateRequest{ + Node: wrapperspb.String("retrying-node"), + Slot: wrapperspb.UInt64(50), + Epoch: wrapperspb.UInt64(1), + StateRoot: wrapperspb.String("0xretry"), + FetchedAt: timestamppb.New(time.Now()), + BeaconImplementation: wrapperspb.String("teku"), + NodeVersion: wrapperspb.String("1.0.0"), + Location: wrapperspb.String(location), + Network: wrapperspb.String("mainnet"), + } + + if _, err := idx.CreateBeaconState(ctx, req); err != nil { + t.Fatalf("first create failed: %v", err) + } + + // A retry of the exact same record (for example, an agent resubmitting + // after a timeout) must still be recognized as AlreadyExists, not + // treated as a location collision. + _, err := idx.CreateBeaconState(ctx, req) + if err == nil { + t.Fatal("expected AlreadyExists for a same-identity retry, got no error") + } + + countRsp, err := idx.CountBeaconState(ctx, &pindexer.CountBeaconStateRequest{}) + if err != nil { + t.Fatalf("failed to count: %v", err) + } + if countRsp.Count.Value != 1 { + t.Fatalf("expected exactly 1 row, got %d", countRsp.Count.Value) + } +} + +func TestCreateExecutionBadBlock_RejectsLocationOwnedByADifferentRecord(t *testing.T) { + idx := newLocationOwnershipTestIndexer(t) + ctx := context.Background() + + location := "execution_bad_block/shared-location.json" + + if _, err := idx.CreateExecutionBadBlock(ctx, &pindexer.CreateExecutionBadBlockRequest{ + Node: wrapperspb.String("owner-node"), + BlockHash: wrapperspb.String("0xowner"), + BlockNumber: wrapperspb.Int64(1), + FetchedAt: timestamppb.New(time.Now()), + Location: wrapperspb.String(location), + ContentEncoding: wrapperspb.String("gzip"), + Network: wrapperspb.String("mainnet"), + ExecutionImplementation: wrapperspb.String("geth"), + NodeVersion: wrapperspb.String("1.0.0"), + }); err != nil { + t.Fatalf("legitimate create failed: %v", err) + } + + // Before this fix, CreateExecutionBadBlock had no dedup or ownership + // check at all, so this collision attempt would have silently + // succeeded. + _, err := idx.CreateExecutionBadBlock(ctx, &pindexer.CreateExecutionBadBlockRequest{ + Node: wrapperspb.String("attacker-node"), + BlockHash: wrapperspb.String("0xdifferent"), + BlockNumber: wrapperspb.Int64(2), + FetchedAt: timestamppb.New(time.Now()), + Location: wrapperspb.String(location), + ContentEncoding: wrapperspb.String("gzip"), + Network: wrapperspb.String("mainnet"), + ExecutionImplementation: wrapperspb.String("geth"), + NodeVersion: wrapperspb.String("1.0.0"), + }) + if err == nil { + t.Fatal("expected the colliding create to be rejected, got no error") + } + + countRsp, err := idx.CountExecutionBadBlock(ctx, &pindexer.CountExecutionBadBlockRequest{}) + if err != nil { + t.Fatalf("failed to count: %v", err) + } + if countRsp.Count.Value != 1 { + t.Fatalf("expected exactly 1 row after the rejected collision attempt, got %d", countRsp.Count.Value) + } +}