JakeHillion/storj
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

satellite/metabase: allow setting metadata with begin object requests

Browse Source 
Turns out that S3 protocol is setting object metadata with initial
request, in our case it's BeginObject, so we need to modify metabase
methods to accept metadata at the beginning and at the end.
BeginObject methods will set metadata always (niled or not niled).

One additional improvment for metadata fields was introduced and it's
validating fields as optional if EncryptedMetadata was not set.

This change currently doesn't have any implications and  it's a base for
other changes. Metadata is not set with BeginObject metainfo endpoint
yet.

Change-Id: I1f768407bc3428500b0d30ee188257420d953001
This commit is contained in:
Michał Niewrzał 2021-10-29 12:38:28 +02:00 committed by Michal Niewrzal
parent bd2448bc4d
commit e792727bea
2 changed files with 220 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

76
satellite/metabase/commit.go
View File

@ -38,21 +38,39 @@ type BeginObjectNextVersion struct {
ExpiresAt *time.Time
ZombieDeletionDeadline *time.Time
EncryptedMetadata []byte // optional
EncryptedMetadataNonce []byte // optional
EncryptedMetadataEncryptedKey []byte // optional
Encryption storj.EncryptionParameters
}
// Verify verifies get object reqest fields.
func (opts *BeginObjectNextVersion) Verify() error {
if err := opts.ObjectStream.Verify(); err != nil {
return err
}
if opts.Version != NextVersion {
return ErrInvalidRequest.New("Version should be metabase.NextVersion")
}
if opts.EncryptedMetadata == nil && (opts.EncryptedMetadataNonce != nil || opts.EncryptedMetadataEncryptedKey != nil) {
return ErrInvalidRequest.New("EncryptedMetadataNonce and EncryptedMetadataEncryptedKey must be empty if EncryptedMetadata is empty")
} else if opts.EncryptedMetadata != nil && (opts.EncryptedMetadataNonce == nil || opts.EncryptedMetadataEncryptedKey == nil) {
return ErrInvalidRequest.New("EncryptedMetadataNonce and EncryptedMetadataEncryptedKey must be not empty if EncryptedMetadata is not empty")
}
return nil
}
// BeginObjectNextVersion adds a pending object to the database, with automatically assigned version.
func (db *DB) BeginObjectNextVersion(ctx context.Context, opts BeginObjectNextVersion) (committed Version, err error) {
defer mon.Task()(&ctx)(&err)
if err := opts.ObjectStream.Verify(); err != nil {
if err := opts.Verify(); err != nil {
return -1, err
}
if opts.Version != NextVersion {
return -1, ErrInvalidRequest.New("Version should be metabase.NextVersion")
}
if opts.ZombieDeletionDeadline == nil {
deadline := time.Now().Add(defaultZombieDeletionPeriod)
opts.ZombieDeletionDeadline = &deadline
@ -62,7 +80,8 @@ func (db *DB) BeginObjectNextVersion(ctx context.Context, opts BeginObjectNextVe
INSERT INTO objects (
project_id, bucket_name, object_key, version, stream_id,
expires_at, encryption,
zombie_deletion_deadline
zombie_deletion_deadline,
encrypted_metadata, encrypted_metadata_nonce, encrypted_metadata_encrypted_key
) VALUES (
$1, $2, $3,
coalesce((
@ -73,11 +92,14 @@ func (db *DB) BeginObjectNextVersion(ctx context.Context, opts BeginObjectNextVe
LIMIT 1
), 1),
$4, $5, $6,
$7)
$7,
$8, $9, $10)
RETURNING version
`, opts.ProjectID, []byte(opts.BucketName), opts.ObjectKey, opts.StreamID,
opts.ExpiresAt, encryptionParameters{&opts.Encryption},
opts.ZombieDeletionDeadline)
opts.ZombieDeletionDeadline,
opts.EncryptedMetadata, opts.EncryptedMetadataNonce, opts.EncryptedMetadataEncryptedKey,
)
var v int64
if err := row.Scan(&v); err != nil {
@ -96,21 +118,39 @@ type BeginObjectExactVersion struct {
ExpiresAt *time.Time
ZombieDeletionDeadline *time.Time
EncryptedMetadata []byte // optional
EncryptedMetadataNonce []byte // optional
EncryptedMetadataEncryptedKey []byte // optional
Encryption storj.EncryptionParameters
}
// Verify verifies get object reqest fields.
func (opts *BeginObjectExactVersion) Verify() error {
if err := opts.ObjectStream.Verify(); err != nil {
return err
}
if opts.Version == NextVersion {
return ErrInvalidRequest.New("Version should not be metabase.NextVersion")
}
if opts.EncryptedMetadata == nil && (opts.EncryptedMetadataNonce != nil || opts.EncryptedMetadataEncryptedKey != nil) {
return ErrInvalidRequest.New("EncryptedMetadataNonce and EncryptedMetadataEncryptedKey must be empty if EncryptedMetadata is empty")
} else if opts.EncryptedMetadata != nil && (opts.EncryptedMetadataNonce == nil || opts.EncryptedMetadataEncryptedKey == nil) {
return ErrInvalidRequest.New("EncryptedMetadataNonce and EncryptedMetadataEncryptedKey must be not empty if EncryptedMetadata is not empty")
}
return nil
}
// BeginObjectExactVersion adds a pending object to the database, with specific version.
func (db *DB) BeginObjectExactVersion(ctx context.Context, opts BeginObjectExactVersion) (committed Object, err error) {
defer mon.Task()(&ctx)(&err)
if err := opts.ObjectStream.Verify(); err != nil {
if err := opts.Verify(); err != nil {
return Object{}, err
}
if opts.Version == NextVersion {
return Object{}, ErrInvalidRequest.New("Version should not be metabase.NextVersion")
}
if opts.ZombieDeletionDeadline == nil {
deadline := time.Now().Add(defaultZombieDeletionPeriod)
opts.ZombieDeletionDeadline = &deadline
@ -133,16 +173,20 @@ func (db *DB) BeginObjectExactVersion(ctx context.Context, opts BeginObjectExact
INSERT INTO objects (
project_id, bucket_name, object_key, version, stream_id,
expires_at, encryption,
zombie_deletion_deadline
zombie_deletion_deadline,
encrypted_metadata, encrypted_metadata_nonce, encrypted_metadata_encrypted_key
) VALUES (
$1, $2, $3, $4, $5,
$6, $7,
$8
$8,
$9, $10, $11
)
RETURNING status, created_at
`, opts.ProjectID, []byte(opts.BucketName), opts.ObjectKey, opts.Version, opts.StreamID,
opts.ExpiresAt, encryptionParameters{&opts.Encryption},
opts.ZombieDeletionDeadline).
opts.ZombieDeletionDeadline,
opts.EncryptedMetadata, opts.EncryptedMetadataNonce, opts.EncryptedMetadataEncryptedKey,
).
Scan(
&object.Status, &object.CreatedAt,
)

166
satellite/metabase/commit_test.go
View File

@ -20,6 +20,13 @@ func TestBeginObjectNextVersion(t *testing.T) {
metabasetest.Run(t, func(ctx *testcontext.Context, t *testing.T, db *metabase.DB) {
obj := metabasetest.RandObjectStream()
objectStream := metabase.ObjectStream{
ProjectID: obj.ProjectID,
BucketName: obj.BucketName,
ObjectKey: obj.ObjectKey,
StreamID: obj.StreamID,
}
for _, test := range metabasetest.InvalidObjectStreams(obj) {
test := test
t.Run(test.Name, func(t *testing.T) {
@ -37,12 +44,33 @@ func TestBeginObjectNextVersion(t *testing.T) {
})
}
objectStream := metabase.ObjectStream{
ProjectID: obj.ProjectID,
BucketName: obj.BucketName,
ObjectKey: obj.ObjectKey,
StreamID: obj.StreamID,
}
t.Run("invalid EncryptedMetadata", func(t *testing.T) {
defer metabasetest.DeleteAll{}.Check(ctx, t, db)
metabasetest.BeginObjectNextVersion{
Opts: metabase.BeginObjectNextVersion{
ObjectStream: objectStream,
Encryption: metabasetest.DefaultEncryption,
EncryptedMetadata: testrand.BytesInt(32),
},
Version: -1,
ErrClass: &metabase.ErrInvalidRequest,
ErrText: "EncryptedMetadataNonce and EncryptedMetadataEncryptedKey must be not empty if EncryptedMetadata is not empty",
}.Check(ctx, t, db)
metabasetest.BeginObjectNextVersion{
Opts: metabase.BeginObjectNextVersion{
ObjectStream: objectStream,
Encryption: metabasetest.DefaultEncryption,
EncryptedMetadataEncryptedKey: testrand.BytesInt(32),
},
Version: -1,
ErrClass: &metabase.ErrInvalidRequest,
ErrText: "EncryptedMetadataNonce and EncryptedMetadataEncryptedKey must be empty if EncryptedMetadata is empty",
}.Check(ctx, t, db)
metabasetest.Verify{}.Check(ctx, t, db)
})
t.Run("disallow exact version", func(t *testing.T) {
defer metabasetest.DeleteAll{}.Check(ctx, t, db)
@ -283,6 +311,54 @@ func TestBeginObjectNextVersion(t *testing.T) {
},
}.Check(ctx, t, db)
})
t.Run("begin object next version with metadata", func(t *testing.T) {
defer metabasetest.DeleteAll{}.Check(ctx, t, db)
now := time.Now()
zombieDeadline := now.Add(24 * time.Hour)
objectStream.Version = metabase.NextVersion
encryptedMetadata := testrand.BytesInt(64)
encryptedMetadataNonce := testrand.Nonce()
encryptedMetadataEncryptedKey := testrand.BytesInt(32)
metabasetest.BeginObjectNextVersion{
Opts: metabase.BeginObjectNextVersion{
ObjectStream: objectStream,
Encryption: metabasetest.DefaultEncryption,
EncryptedMetadata: encryptedMetadata,
EncryptedMetadataNonce: encryptedMetadataNonce[:],
EncryptedMetadataEncryptedKey: encryptedMetadataEncryptedKey,
},
Version: 1,
}.Check(ctx, t, db)
metabasetest.Verify{
Objects: []metabase.RawObject{
{
ObjectStream: metabase.ObjectStream{
ProjectID: obj.ProjectID,
BucketName: obj.BucketName,
ObjectKey: obj.ObjectKey,
Version: metabase.DefaultVersion,
StreamID: obj.StreamID,
},
CreatedAt: now,
Status: metabase.Pending,
EncryptedMetadata: encryptedMetadata,
EncryptedMetadataNonce: encryptedMetadataNonce[:],
EncryptedMetadataEncryptedKey: encryptedMetadataEncryptedKey,
Encryption: metabasetest.DefaultEncryption,
ZombieDeletionDeadline: &zombieDeadline,
},
},
}.Check(ctx, t, db)
})
})
}
@ -314,6 +390,36 @@ func TestBeginObjectExactVersion(t *testing.T) {
StreamID: obj.StreamID,
}
t.Run("invalid EncryptedMetadata", func(t *testing.T) {
defer metabasetest.DeleteAll{}.Check(ctx, t, db)
objectStream.Version = 1
metabasetest.BeginObjectExactVersion{
Opts: metabase.BeginObjectExactVersion{
ObjectStream: objectStream,
Encryption: metabasetest.DefaultEncryption,
EncryptedMetadata: testrand.BytesInt(32),
},
Version: metabase.DefaultVersion,
ErrClass: &metabase.ErrInvalidRequest,
ErrText: "EncryptedMetadataNonce and EncryptedMetadataEncryptedKey must be not empty if EncryptedMetadata is not empty",
}.Check(ctx, t, db)
metabasetest.BeginObjectExactVersion{
Opts: metabase.BeginObjectExactVersion{
ObjectStream: objectStream,
Encryption: metabasetest.DefaultEncryption,
EncryptedMetadataEncryptedKey: testrand.BytesInt(32),
},
Version: metabase.DefaultVersion,
ErrClass: &metabase.ErrInvalidRequest,
ErrText: "EncryptedMetadataNonce and EncryptedMetadataEncryptedKey must be empty if EncryptedMetadata is empty",
}.Check(ctx, t, db)
metabasetest.Verify{}.Check(ctx, t, db)
})
t.Run("disallow NextVersion", func(t *testing.T) {
defer metabasetest.DeleteAll{}.Check(ctx, t, db)
@ -601,6 +707,54 @@ func TestBeginObjectExactVersion(t *testing.T) {
},
}.Check(ctx, t, db)
})
t.Run("begin object exact version with metadata", func(t *testing.T) {
defer metabasetest.DeleteAll{}.Check(ctx, t, db)
now := time.Now()
zombieDeadline := now.Add(24 * time.Hour)
objectStream.Version = 1
encryptedMetadata := testrand.BytesInt(64)
encryptedMetadataNonce := testrand.Nonce()
encryptedMetadataEncryptedKey := testrand.BytesInt(32)
metabasetest.BeginObjectExactVersion{
Opts: metabase.BeginObjectExactVersion{
ObjectStream: objectStream,
Encryption: metabasetest.DefaultEncryption,
EncryptedMetadata: encryptedMetadata,
EncryptedMetadataNonce: encryptedMetadataNonce[:],
EncryptedMetadataEncryptedKey: encryptedMetadataEncryptedKey,
},
Version: 1,
}.Check(ctx, t, db)
metabasetest.Verify{
Objects: []metabase.RawObject{
{
ObjectStream: metabase.ObjectStream{
ProjectID: obj.ProjectID,
BucketName: obj.BucketName,
ObjectKey: obj.ObjectKey,
Version: metabase.DefaultVersion,
StreamID: obj.StreamID,
},
CreatedAt: now,
Status: metabase.Pending,
EncryptedMetadata: encryptedMetadata,
EncryptedMetadataNonce: encryptedMetadataNonce[:],
EncryptedMetadataEncryptedKey: encryptedMetadataEncryptedKey,
Encryption: metabasetest.DefaultEncryption,
ZombieDeletionDeadline: &zombieDeadline,
},
},
}.Check(ctx, t, db)
})
})
}