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

satellite/metainfo/metabase: cast bucket_name to []byte

Browse Source 
Currently our metabase assumption is that it may contain arbitrary
bucket names and endpoint applies the naming constraints as it sees fit.
However by passing bucket_name as TEXT pg and crdb automatically try to
convert it to []byte, which may or not may work as intended... or in
some cases not work at all.

Cast all bucket name arguments to []byte to make it work.

Change-Id: I44650f5c873010997398bb0163d7f56ff6d9b5cf
This commit is contained in:
Egon Elbre 2021-03-02 13:22:49 +02:00
parent ba0197a9b7
commit 7e72a231c2
7 changed files with 30 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
satellite/metainfo/metabase/commit.go
View File

@ -67,7 +67,7 @@ func (db *DB) BeginObjectNextVersion(ctx context.Context, opts BeginObjectNextVe
$4, $5, $6,
$7)
RETURNING version
`, opts.ProjectID, opts.BucketName, []byte(opts.ObjectKey), opts.StreamID,
`, opts.ProjectID, []byte(opts.BucketName), []byte(opts.ObjectKey), opts.StreamID,
opts.ExpiresAt, encryptionParameters{&opts.Encryption},
opts.ZombieDeletionDeadline)
@ -125,7 +125,7 @@ func (db *DB) BeginObjectExactVersion(ctx context.Context, opts BeginObjectExact
$8
)
RETURNING status, created_at
`, opts.ProjectID, opts.BucketName, []byte(opts.ObjectKey), opts.Version, opts.StreamID,
`, opts.ProjectID, []byte(opts.BucketName), []byte(opts.ObjectKey), opts.Version, opts.StreamID,
opts.ExpiresAt, encryptionParameters{&opts.Encryption},
opts.ZombieDeletionDeadline).
Scan(
@ -183,7 +183,7 @@ func (db *DB) BeginSegment(ctx context.Context, opts BeginSegment) (err error) {
version = $4 AND
stream_id = $5 AND
status = `+pendingStatus,
opts.ProjectID, opts.BucketName, []byte(opts.ObjectKey), opts.Version, opts.StreamID).Scan(&value)
opts.ProjectID, []byte(opts.BucketName), []byte(opts.ObjectKey), opts.Version, opts.StreamID).Scan(&value)
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
return Error.New("pending object missing")
@ -289,7 +289,7 @@ func (db *DB) CommitSegment(ctx context.Context, opts CommitSegment) (err error)
opts.EncryptedSize, opts.PlainOffset, opts.PlainSize,
redundancyScheme{&opts.Redundancy},
aliasPieces,
opts.ProjectID, opts.BucketName, []byte(opts.ObjectKey), opts.Version, opts.StreamID,
opts.ProjectID, []byte(opts.BucketName), []byte(opts.ObjectKey), opts.Version, opts.StreamID,
)
if err != nil {
if code := pgerrcode.FromError(err); code == pgxerrcode.NotNullViolation {
@ -364,7 +364,7 @@ func (db *DB) CommitInlineSegment(ctx context.Context, opts CommitInlineSegment)
storj.PieceID{}, opts.EncryptedKeyNonce, opts.EncryptedKey,
len(opts.InlineData), opts.PlainOffset, opts.PlainSize,
opts.InlineData,
opts.ProjectID, opts.BucketName, []byte(opts.ObjectKey), opts.Version, opts.StreamID,
opts.ProjectID, []byte(opts.BucketName), []byte(opts.ObjectKey), opts.Version, opts.StreamID,
)
if err != nil {
if code := pgerrcode.FromError(err); code == pgxerrcode.NotNullViolation {
@ -502,7 +502,7 @@ func (db *DB) CommitObject(ctx context.Context, opts CommitObject) (object Objec
RETURNING
created_at, expires_at,
encryption;
`, opts.ProjectID, opts.BucketName, []byte(opts.ObjectKey), opts.Version, opts.StreamID,
`, opts.ProjectID, []byte(opts.BucketName), []byte(opts.ObjectKey), opts.Version, opts.StreamID,
len(segments),
opts.EncryptedMetadataNonce, opts.EncryptedMetadata, opts.EncryptedMetadataEncryptedKey,
totalPlainSize,
@ -583,7 +583,7 @@ func (db *DB) UpdateObjectMetadata(ctx context.Context, opts UpdateObjectMetadat
version = $4 AND
stream_id = $5 AND
status = `+committedStatus,
opts.ProjectID, opts.BucketName, []byte(opts.ObjectKey), opts.Version, opts.StreamID,
opts.ProjectID, []byte(opts.BucketName), []byte(opts.ObjectKey), opts.Version, opts.StreamID,
opts.EncryptedMetadataNonce, opts.EncryptedMetadata, opts.EncryptedMetadataEncryptedKey)
if err != nil {
return Error.New("unable to update object metadata: %w", err)

2
satellite/metainfo/metabase/commit_object.go
View File

@ -110,7 +110,7 @@ func (db *DB) CommitObjectWithSegments(ctx context.Context, opts CommitObjectWit
RETURNING
created_at, expires_at,
encryption;
`, opts.ProjectID, opts.BucketName, []byte(opts.ObjectKey), opts.Version, opts.StreamID,
`, opts.ProjectID, []byte(opts.BucketName), []byte(opts.ObjectKey), opts.Version, opts.StreamID,
len(finalSegments),
opts.EncryptedMetadataNonce, opts.EncryptedMetadata, opts.EncryptedMetadataEncryptedKey,
totalPlainSize,

12
satellite/metainfo/metabase/delete.go
View File

@ -118,7 +118,7 @@ func (db *DB) DeleteObjectExactVersion(ctx context.Context, opts DeleteObjectExa
encrypted_metadata_nonce, encrypted_metadata, encrypted_metadata_encrypted_key,
total_plain_size, total_encrypted_size, fixed_segment_size,
encryption;
`, opts.ProjectID, opts.BucketName, []byte(opts.ObjectKey), opts.Version)
`, opts.ProjectID, []byte(opts.BucketName), []byte(opts.ObjectKey), opts.Version)
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
return storj.ErrObjectNotFound.Wrap(Error.Wrap(err))
@ -198,7 +198,7 @@ func (db *DB) DeletePendingObject(ctx context.Context, opts DeletePendingObject)
encrypted_metadata_nonce, encrypted_metadata, encrypted_metadata_encrypted_key,
total_plain_size, total_encrypted_size, fixed_segment_size,
encryption;
`, opts.ProjectID, opts.BucketName, []byte(opts.ObjectKey), opts.Version, opts.StreamID)
`, opts.ProjectID, []byte(opts.BucketName), []byte(opts.ObjectKey), opts.Version, opts.StreamID)
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
return storj.ErrObjectNotFound.Wrap(Error.Wrap(err))
@ -253,7 +253,7 @@ func (db *DB) DeleteObjectLatestVersion(ctx context.Context, opts DeleteObjectLa
// ORDER BY version DESC
// LIMIT 1
// RETURNING stream_id;
// `, opts.ProjectID, opts.BucketName, opts.ObjectKey)
// `, opts.ProjectID, []byte(opts.BucketName), opts.ObjectKey)
// version for Postgres and Cockroachdb (but slow for Cockroachdb)
rows, err := tx.Query(ctx, `
@ -277,7 +277,7 @@ func (db *DB) DeleteObjectLatestVersion(ctx context.Context, opts DeleteObjectLa
encrypted_metadata_nonce, encrypted_metadata, encrypted_metadata_encrypted_key,
total_plain_size, total_encrypted_size, fixed_segment_size,
encryption;
`, opts.ProjectID, opts.BucketName, []byte(opts.ObjectKey))
`, opts.ProjectID, []byte(opts.BucketName), []byte(opts.ObjectKey))
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
return storj.ErrObjectNotFound.Wrap(Error.Wrap(err))
@ -332,7 +332,7 @@ func (db *DB) DeleteObjectAnyStatusAllVersions(ctx context.Context, opts DeleteO
encrypted_metadata_nonce, encrypted_metadata, encrypted_metadata_encrypted_key,
total_plain_size, total_encrypted_size, fixed_segment_size,
encryption;
`, opts.ProjectID, opts.BucketName, []byte(opts.ObjectKey))
`, opts.ProjectID, []byte(opts.BucketName), []byte(opts.ObjectKey))
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
return storj.ErrObjectNotFound.Wrap(Error.Wrap(err))
@ -410,7 +410,7 @@ func (db *DB) DeleteObjectsAllVersions(ctx context.Context, opts DeleteObjectsAl
encrypted_metadata_nonce, encrypted_metadata, encrypted_metadata_encrypted_key,
total_plain_size, total_encrypted_size, fixed_segment_size,
encryption;
`, projectID, bucketName, pgutil.ByteaArray(objectKeys))
`, projectID, []byte(bucketName), pgutil.ByteaArray(objectKeys))
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
return storj.ErrObjectNotFound.Wrap(Error.Wrap(err))

2
satellite/metainfo/metabase/delete_bucket.go
View File

@ -76,7 +76,7 @@ func (db *DB) DeleteBucketObjects(ctx context.Context, opts DeleteBucketObjects)
for {
deleteSegments = deleteSegments[:0]
err = withRows(db.db.Query(ctx, query,
opts.Bucket.ProjectID, opts.Bucket.BucketName, batchSize))(func(rows tagsql.Rows) error {
opts.Bucket.ProjectID, []byte(opts.Bucket.BucketName), batchSize))(func(rows tagsql.Rows) error {
ids := map[uuid.UUID]struct{}{} // TODO: avoid map here
for rows.Next() {
var streamID uuid.UUID

12
satellite/metainfo/metabase/get.go
View File

@ -73,7 +73,7 @@ func (db *DB) GetObjectExactVersion(ctx context.Context, opts GetObjectExactVers
object_key = $3 AND
version = $4 AND
status = `+committedStatus,
opts.ProjectID, opts.BucketName, []byte(opts.ObjectKey), opts.Version).
opts.ProjectID, []byte(opts.BucketName), []byte(opts.ObjectKey), opts.Version).
Scan(
&object.StreamID,
&object.CreatedAt, &object.ExpiresAt,
@ -130,7 +130,7 @@ func (db *DB) GetObjectLatestVersion(ctx context.Context, opts GetObjectLatestVe
status = `+committedStatus+`
ORDER BY version desc
LIMIT 1
`, opts.ProjectID, opts.BucketName, []byte(opts.ObjectKey)).
`, opts.ProjectID, []byte(opts.BucketName), []byte(opts.ObjectKey)).
Scan(
&object.StreamID, &object.Version,
&object.CreatedAt, &object.ExpiresAt,
@ -186,7 +186,7 @@ func (db *DB) GetSegmentByLocation(ctx context.Context, opts GetSegmentByLocatio
LIMIT 1
) AND
position = $4
`, opts.ProjectID, opts.BucketName, []byte(opts.ObjectKey), opts.Position.Encode()).
`, opts.ProjectID, []byte(opts.BucketName), []byte(opts.ObjectKey), opts.Position.Encode()).
Scan(
&segment.StreamID,
&segment.RootPieceID, &segment.EncryptedKeyNonce, &segment.EncryptedKey,
@ -301,7 +301,7 @@ func (db *DB) GetLatestObjectLastSegment(ctx context.Context, opts GetLatestObje
)
ORDER BY position DESC
LIMIT 1
`, opts.ProjectID, opts.BucketName, []byte(opts.ObjectKey)).
`, opts.ProjectID, []byte(opts.BucketName), []byte(opts.ObjectKey)).
Scan(
&segment.StreamID, &segment.Position,
&segment.RootPieceID, &segment.EncryptedKeyNonce, &segment.EncryptedKey,
@ -364,7 +364,7 @@ func (db *DB) GetSegmentByOffset(ctx context.Context, opts GetSegmentByOffset) (
(plain_size + plain_offset) > $4
ORDER BY plain_offset ASC
LIMIT 1
`, opts.ProjectID, opts.BucketName, []byte(opts.ObjectKey), opts.PlainOffset).
`, opts.ProjectID, []byte(opts.BucketName), []byte(opts.ObjectKey), opts.PlainOffset).
Scan(
&segment.StreamID, &segment.Position,
&segment.RootPieceID, &segment.EncryptedKeyNonce, &segment.EncryptedKey,
@ -414,7 +414,7 @@ func (db *DB) BucketEmpty(ctx context.Context, opts BucketEmpty) (empty bool, er
project_id = $1 AND
bucket_name = $2
LIMIT 1
`, opts.ProjectID, opts.BucketName).Scan(&value)
`, opts.ProjectID, []byte(opts.BucketName)).Scan(&value)
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
return true, nil

14
satellite/metainfo/metabase/iterator.go
View File

@ -19,7 +19,7 @@ type objectsIterator struct {
db *DB
projectID uuid.UUID
bucketName string
bucketName []byte
status ObjectStatus
prefix ObjectKey
prefixLimit ObjectKey
@ -49,7 +49,7 @@ func iterateAllVersions(ctx context.Context, db *DB, opts IterateObjects, fn fun
db: db,
projectID: opts.ProjectID,
bucketName: opts.BucketName,
bucketName: []byte(opts.BucketName),
prefix: opts.Prefix,
prefixLimit: prefixLimit(opts.Prefix),
batchSize: opts.BatchSize,
@ -82,7 +82,7 @@ func iterateAllVersionsWithStatus(ctx context.Context, db *DB, opts IterateObjec
db: db,
projectID: opts.ProjectID,
bucketName: opts.BucketName,
bucketName: []byte(opts.BucketName),
status: opts.Status,
prefix: opts.Prefix,
prefixLimit: prefixLimit(opts.Prefix),
@ -121,7 +121,7 @@ func iteratePendingObjectsByKey(ctx context.Context, db *DB, opts IteratePending
db: db,
projectID: opts.ProjectID,
bucketName: opts.BucketName,
bucketName: []byte(opts.BucketName),
prefix: "",
prefixLimit: "",
batchSize: opts.BatchSize,
@ -366,8 +366,10 @@ func doNextQueryAllVersionsWithStatus(ctx context.Context, it *objectsIterator)
}
// nextBucket returns the lexicographically next bucket.
func nextBucket(b string) string {
return b + "\x01"
func nextBucket(b []byte) []byte {
xs := make([]byte, len(b)+1)
copy(xs, b)
return xs
}
// doNextQuery executes query to fetch the next batch returning the rows.

2
satellite/metainfo/metabase/loop.go
View File

@ -174,7 +174,7 @@ func (it *loopIterator) doNextQuery(ctx context.Context) (_ tagsql.Rows, err err
WHERE (project_id, bucket_name, object_key, version) > ($1, $2, $3, $4)
ORDER BY project_id ASC, bucket_name ASC, object_key ASC, version ASC
LIMIT $5
`, it.cursor.ProjectID, it.cursor.BucketName,
`, it.cursor.ProjectID, []byte(it.cursor.BucketName),
[]byte(it.cursor.ObjectKey), int(it.cursor.Version),
it.batchSize,
)