satellite/metainfo: limit size of uplink-provided metadata to 2KiB

Change-Id: Id44a46046ddb4a12102525531f4502fcff2b6252

private/testplanet/reconfigure.go

@@ -76,3 +76,10 @@ var MaxSegmentSize = func(maxSegmentSize memory.Size) func(log *zap.Logger, inde
 		config.Metainfo.MaxSegmentSize = maxSegmentSize
 	}
 }
+
+// MaxMetadataSize returns function to change satellite max metadata size value.
+var MaxMetadataSize = func(maxMetadataSize memory.Size) func(log *zap.Logger, index int, config *satellite.Config) {
+	return func(log *zap.Logger, index int, config *satellite.Config) {
+		config.Metainfo.MaxMetadataSize = maxMetadataSize
+	}
+}

private/testplanet/satellite.go

@@ -423,6 +423,7 @@ func (planet *Planet) newSatellites(count int, satelliteDatabases satellitedbtes
 				MinRemoteSegmentSize: 0,  // TODO: fix tests to work with 1024
 				MaxInlineSegmentSize: 4 * memory.KiB,
 				MaxSegmentSize:       64 * memory.MiB,
+				MaxMetadataSize:      2 * memory.KiB,
 				MaxCommitInterval:    1 * time.Hour,
 				Overlay:              true,
 				RS: metainfo.RSConfig{

satellite/metainfo/config.go

@@ -52,6 +52,7 @@ type Config struct {
 	MinRemoteSegmentSize memory.Size          `default:"1240" help:"minimum remote segment size"`
 	MaxInlineSegmentSize memory.Size          `default:"4KiB" help:"maximum inline segment size"`
 	MaxSegmentSize       memory.Size          `default:"64MiB" help:"maximum segment size"`
+	MaxMetadataSize      memory.Size          `default:"2KiB" help:"maximum segment metadata size"`
 	MaxCommitInterval    time.Duration        `default:"48h" help:"maximum time allowed to pass between creating and committing a segment"`
 	Overlay              bool                 `default:"true" help:"toggle flag if overlay is enabled"`
 	RS                   RSConfig             `help:"redundancy scheme configuration"`

satellite/metainfo/metainfo.go

@@ -19,6 +19,7 @@ import (
 	"storj.io/common/encryption"
 	"storj.io/common/errs2"
 	"storj.io/common/macaroon"
+	"storj.io/common/memory"
 	"storj.io/common/pb"
 	"storj.io/common/rpc/rpcstatus"
 	"storj.io/common/signing"
@@ -1001,6 +1002,10 @@ func (endpoint *Endpoint) commitObject(ctx context.Context, req *pb.ObjectCommit
 		return nil, err
 	}
 
+	metadataSize := memory.Size(len(req.EncryptedMetadata))
+	if metadataSize > endpoint.config.MaxMetadataSize {
+		return nil, rpcstatus.Error(rpcstatus.InvalidArgument, fmt.Sprintf("Metadata is too large, got %v, maximum allowed is %v", metadataSize, endpoint.config.MaxMetadataSize))
+	}
 	streamMeta := pb.StreamMeta{}
 	err = pb.Unmarshal(req.EncryptedMetadata, &streamMeta)
 	if err != nil {

satellite/metainfo/metainfo_test.go

@@ -1515,3 +1515,120 @@ func TestInlineSegmentThreshold(t *testing.T) {
 		}
 	})
 }
+
+// TestCommitObjectMetadataSize ensures that CommitObject returns an error when the metadata provided by the user is too large.
+func TestCommitObjectMetadataSize(t *testing.T) {
+	testplanet.Run(t, testplanet.Config{
+		SatelliteCount: 1, StorageNodeCount: 4, UplinkCount: 1,
+		Reconfigure: testplanet.Reconfigure{
+			Satellite: testplanet.MaxMetadataSize(2 * memory.KiB),
+		},
+	}, func(t *testing.T, ctx *testcontext.Context, planet *testplanet.Planet) {
+		apiKey := planet.Uplinks[0].APIKey[planet.Satellites[0].ID()]
+		metainfoService := planet.Satellites[0].Metainfo.Service
+
+		projects, err := planet.Satellites[0].DB.Console().Projects().GetAll(ctx)
+		require.NoError(t, err)
+		projectID := projects[0].ID
+
+		bucket := storj.Bucket{
+			Name:      "initial-bucket",
+			ProjectID: projectID,
+		}
+		_, err = metainfoService.CreateBucket(ctx, bucket)
+		require.NoError(t, err)
+
+		metainfoClient, err := planet.Uplinks[0].DialMetainfo(ctx, planet.Satellites[0], apiKey)
+		require.NoError(t, err)
+		defer ctx.Check(metainfoClient.Close)
+
+		params := metainfo.BeginObjectParams{
+			Bucket:        []byte(bucket.Name),
+			EncryptedPath: []byte("encrypted-path"),
+			Redundancy: storj.RedundancyScheme{
+				Algorithm:      storj.ReedSolomon,
+				ShareSize:      256,
+				RequiredShares: 1,
+				RepairShares:   1,
+				OptimalShares:  3,
+				TotalShares:    4,
+			},
+			EncryptionParameters: storj.EncryptionParameters{},
+			ExpiresAt:            time.Now().Add(24 * time.Hour),
+		}
+		beginObjectResponse, err := metainfoClient.BeginObject(ctx, params)
+		require.NoError(t, err)
+
+		segmentID, limits, _, err := metainfoClient.BeginSegment(ctx, metainfo.BeginSegmentParams{
+			StreamID: beginObjectResponse.StreamID,
+			Position: storj.SegmentPosition{
+				Index: 0,
+			},
+			MaxOrderLimit: memory.MiB.Int64(),
+		})
+		require.NoError(t, err)
+
+		fullIDMap := make(map[storj.NodeID]*identity.FullIdentity)
+		for _, node := range planet.StorageNodes {
+			fullIDMap[node.ID()] = node.Identity
+		}
+
+		makeResult := func(num int32) *pb.SegmentPieceUploadResult {
+			nodeID := limits[num].Limit.StorageNodeId
+			hash := &pb.PieceHash{
+				PieceId:   limits[num].Limit.PieceId,
+				PieceSize: 1048832,
+				Timestamp: time.Now(),
+			}
+
+			fullID := fullIDMap[nodeID]
+			require.NotNil(t, fullID)
+			signer := signing.SignerFromFullIdentity(fullID)
+			signedHash, err := signing.SignPieceHash(ctx, signer, hash)
+			require.NoError(t, err)
+
+			return &pb.SegmentPieceUploadResult{
+				PieceNum: num,
+				NodeId:   nodeID,
+				Hash:     signedHash,
+			}
+		}
+		err = metainfoClient.CommitSegment(ctx, metainfo.CommitSegmentParams{
+			SegmentID: segmentID,
+
+			SizeEncryptedData: memory.MiB.Int64(),
+			UploadResult: []*pb.SegmentPieceUploadResult{
+				makeResult(0),
+				makeResult(1),
+				makeResult(2),
+			},
+		})
+		require.NoError(t, err)
+
+		// 5KiB metadata should fail because it is too large.
+		metadata, err := pb.Marshal(&pb.StreamMeta{
+			EncryptedStreamInfo: testrand.Bytes(5 * memory.KiB),
+			NumberOfSegments:    1,
+		})
+		require.NoError(t, err)
+		err = metainfoClient.CommitObject(ctx, metainfo.CommitObjectParams{
+			StreamID:          beginObjectResponse.StreamID,
+			EncryptedMetadata: metadata,
+		})
+		require.Error(t, err)
+		assertInvalidArgument(t, err, true)
+
+		// 1KiB metadata should not fail.
+		metadata, err = pb.Marshal(&pb.StreamMeta{
+			EncryptedStreamInfo: testrand.Bytes(1 * memory.KiB),
+			NumberOfSegments:    1,
+		})
+		require.NoError(t, err)
+		err = metainfoClient.CommitObject(ctx, metainfo.CommitObjectParams{
+			StreamID:          beginObjectResponse.StreamID,
+			EncryptedMetadata: metadata,
+		})
+		require.NoError(t, err)
+	})
+
+}

scripts/testdata/satellite-config.yaml.lock

@@ -316,6 +316,9 @@ identity.key-path: /root/.local/share/storj/identity/satellite/identity.key
 # maximum inline segment size
 # metainfo.max-inline-segment-size: 4.0 KiB
 
+# maximum segment metadata size
+# metainfo.max-metadata-size: 2.0 KiB
+
 # maximum segment size
 # metainfo.max-segment-size: 64.0 MiB