objstore: add experimental encryption wrapper

Signed-off-by: Michael Hoffmann <[email protected]>

CHANGELOG.md

@@ -9,6 +9,7 @@ NOTE: As semantic versioning states all 0.y.z releases can contain breaking chan
 We use *breaking :warning:* to mark changes that are not backward compatible (relates only to v0.y.z releases.)
 
 ## Unreleased
+- [#46](https://github.com/thanos-io/objstore/pull/46) Objstore: Add experimental encryption wrapper
 
 ### Fixed
 - [#33](https://github.com/thanos-io/objstore/pull/33) Tracing: Add `ContextWithTracer()` to inject the tracer into the context.

README.md

@@ -49,6 +49,10 @@ See [MAINTAINERS.md](https://github.com/thanos-io/thanos/blob/main/MAINTAINERS.m
 The core this module is the [`Bucket` interface](objstore.go):
 
 ```go mdox-exec="sed -n '37,50p' objstore.go"
+	OpDelete     = "delete"
+	OpAttributes = "attributes"
+)
+
 // Bucket provides read and write access to an object storage bucket.
 // NOTE: We assume strong consistency for write-read flow.
 type Bucket interface {
@@ -59,15 +63,15 @@ type Bucket interface {
 	// Upload should be idempotent.
 	Upload(ctx context.Context, name string, r io.Reader) error
 
-	// Delete removes the object with the given name.
-	// If object does not exists in the moment of deletion, Delete should throw error.
-	Delete(ctx context.Context, name string) error
-
 ```
 
 All [provider implementations](providers) have to implement `Bucket` interface that allows common read and write operations that all supported by all object providers. If you want to limit the code that will do bucket operation to only read access (smart idea, allowing to limit access permissions), you can use the [`BucketReader` interface](objstore.go):
 
 ```go mdox-exec="sed -n '68,88p' objstore.go"
+	// thanos_objstore_bucket_operation_failures_total metric.
+	// TODO(bwplotka): Remove this when moved to Go 1.14 and replace with InstrumentedBucketReader.
+	ReaderWithExpectedErrs(IsOpFailureExpectedFunc) BucketReader
+}
 
 // BucketReader provides read access to an object storage bucket.
 type BucketReader interface {
@@ -85,10 +89,6 @@ type BucketReader interface {
 	// Exists checks if the given object exists in the bucket.
 	Exists(ctx context.Context, name string) (bool, error)
 
-	// IsObjNotFoundErr returns true if error means that object is not found. Relevant to Get operations.
-	IsObjNotFoundErr(err error) bool
-
-	// Attributes returns information about the specified object.
 ```
 
 Those interfaces represent the object storage operations your code can use from `objstore` clients.
@@ -152,6 +152,7 @@ config:
   insecure: false
   signature_version2: false
   secret_key: ""
+  session_token: ""
   put_user_metadata: {}
   http_config:
     idle_conn_timeout: 1m30s
@@ -181,6 +182,9 @@ config:
     encryption_key: ""
   sts_endpoint: ""
 prefix: ""
+client_side_encryption:
+  enabled: false
+  key_base64: ""
 ```
 
 At a minimum, you will need to provide a value for the `bucket`, `endpoint`, `access_key`, and `secret_key` keys. The rest of the keys are optional.
@@ -345,6 +349,9 @@ config:
   bucket: ""
   service_account: ""
 prefix: ""
+client_side_encryption:
+  enabled: false
+  key_base64: ""
 ```
 
 ###### Using GOOGLE_APPLICATION_CREDENTIALS
@@ -445,6 +452,9 @@ config:
     disable_compression: false
   msi_resource: ""
 prefix: ""
+client_side_encryption:
+  enabled: false
+  key_base64: ""
 ```
 
 If `msi_resource` is used, authentication is done via system-assigned managed identity. The value for Azure should be `https://<storage-account-name>.blob.core.windows.net`.
@@ -489,6 +499,9 @@ config:
   timeout: 5m
   use_dynamic_large_objects: false
 prefix: ""
+client_side_encryption:
+  enabled: false
+  key_base64: ""
 ```
 
 ##### Tencent COS
@@ -523,6 +536,9 @@ config:
       insecure_skip_verify: false
     disable_compression: false
 prefix: ""
+client_side_encryption:
+  enabled: false
+  key_base64: ""
 ```
 
 The `secret_key` and `secret_id` field is required. The `http_config` field is optional for optimize HTTP transport settings. There are two ways to configure the required bucket information:
@@ -543,6 +559,9 @@ config:
   access_key_id: ""
   access_key_secret: ""
 prefix: ""
+client_side_encryption:
+  enabled: false
+  key_base64: ""
 ```
 
 ##### Baidu BOS
@@ -557,6 +576,9 @@ config:
   access_key: ""
   secret_key: ""
 prefix: ""
+client_side_encryption:
+  enabled: false
+  key_base64: ""
 ```
 
 ##### Filesystem
@@ -572,6 +594,9 @@ type: FILESYSTEM
 config:
   directory: ""
 prefix: ""
+client_side_encryption:
+  enabled: false
+  key_base64: ""
 ```
 
 ### Oracle Cloud Infrastructure Object Storage

client/factory.go

@@ -5,6 +5,7 @@ package client
 
 import (
 	"context"
+	"encoding/base64"
 	"fmt"
 	"strings"
 
@@ -41,9 +42,15 @@ const (
 )
 
 type BucketConfig struct {
-	Type   ObjProvider `yaml:"type"`
-	Config interface{} `yaml:"config"`
-	Prefix string      `yaml:"prefix" default:""`
+	Type                 ObjProvider                `yaml:"type"`
+	Config               interface{}                `yaml:"config"`
+	Prefix               string                     `yaml:"prefix" default:""`
+	ClientSideEncryption ClientSideEncryptionConfig `yaml:"client_side_encryption"`
+}
+
+type ClientSideEncryptionConfig struct {
+	Enabled   bool   `yaml:"enabled"`
+	KeyBase64 string `yaml:"key_base64"`
 }
 
 // NewBucket initializes and returns new object storage clients.
@@ -87,5 +94,16 @@ func NewBucket(logger log.Logger, confContentYaml []byte, reg prometheus.Registe
 		return nil, errors.Wrap(err, fmt.Sprintf("create %s client", bucketConf.Type))
 	}
 
+	if bucketConf.ClientSideEncryption.Enabled {
+		key, err := base64.RawStdEncoding.DecodeString(bucketConf.ClientSideEncryption.KeyBase64)
+		if err != nil {
+			return nil, errors.Wrap(err, "unable to read base64 key")
+		}
+		if len(key) != 32 {
+			return nil, errors.New("decoded key must have size 32")
+		}
+		bucket = objstore.BucketWithEncryption(bucket, key)
+	}
+
 	return objstore.NewTracingBucket(objstore.BucketWithMetrics(bucket.Name(), objstore.NewPrefixedBucket(bucket, bucketConf.Prefix), reg)), nil
 }

go.mod

@@ -13,6 +13,7 @@ require (
 	github.com/fatih/structtag v1.2.0
 	github.com/go-kit/log v0.2.1
 	github.com/minio/minio-go/v7 v7.0.45
+	github.com/minio/sio v0.3.0
 	github.com/ncw/swift v1.0.53
 	github.com/opentracing/opentracing-go v1.2.0
 	github.com/oracle/oci-go-sdk/v65 v65.13.0
@@ -100,5 +101,5 @@ require (
 	github.com/Azure/azure-sdk-for-go/sdk/storage/azblob v0.5.1
 	github.com/kr/text v0.2.0 // indirect
 	github.com/satori/go.uuid v1.2.1-0.20181028125025-b2ce2384e17b // indirect
-	golang.org/x/crypto v0.3.0 // indirect
+	golang.org/x/crypto v0.6.0 // indirect
 )

go.sum

@@ -307,6 +307,8 @@ github.com/minio/minio-go/v7 v7.0.45 h1:g4IeM9M9pW/Lo8AGGNOjBZYlvmtlE1N5TQEYWXRW
 github.com/minio/minio-go/v7 v7.0.45/go.mod h1:nCrRzjoSUQh8hgKKtu3Y708OLvRLtuASMg2/nvmbarw=
 github.com/minio/sha256-simd v1.0.0 h1:v1ta+49hkWZyvaKwrQB8elexRqm6Y0aMLjCNsrYxo6g=
 github.com/minio/sha256-simd v1.0.0/go.mod h1:OuYzVNI5vcoYIAmbIvHPl3N3jUzVedXbKy5RFepssQM=
+github.com/minio/sio v0.3.0 h1:syEFBewzOMOYVzSTFpp1MqpSZk8rUNbz8VIIc+PNzus=
+github.com/minio/sio v0.3.0/go.mod h1:8b0yPp2avGThviy/+OCJBI6OMpvxoUuiLvE6F1lebhw=
 github.com/mitchellh/mapstructure v1.4.3 h1:OVowDSCllw/YjdLkam3/sm7wEtOy59d8ndGgCcyj8cs=
 github.com/mitchellh/mapstructure v1.4.3/go.mod h1:bFUtVrKA4DC2yAKiSyO/QUcy7e+RRV2QTWOzhPopBRo=
 github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
@@ -407,11 +409,12 @@ go.uber.org/atomic v1.9.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
 golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20190510104115-cbcb75029529/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
+golang.org/x/crypto v0.0.0-20190513172903-22d7a77e9e5f/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20190605123033-f99c8df09eb5/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
-golang.org/x/crypto v0.3.0 h1:a06MkbcxBrEFc0w0QIZWXrH/9cCX6KJyWbBOIwAn+7A=
-golang.org/x/crypto v0.3.0/go.mod h1:hebNnKkNXi2UzZN1eVRvBB7co0a+JxK6XbPiWVs/3J4=
+golang.org/x/crypto v0.6.0 h1:qfktjS5LUO+fFKeJXZ+ikTRijMmljikvG68fpMMruSc=
+golang.org/x/crypto v0.6.0/go.mod h1:OFC/31mSvZgRz0V1QTNCzfAI1aIRzbiufJtkMIlEp58=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20190306152737-a1d7652674e8/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20190510132918-efd6b22b2522/go.mod h1:ZjyILWgesfNpC6sMxTJOJm9Kp84zZh5NQWvqDGG3Qr8=

objstore.go

@@ -6,8 +6,12 @@ package objstore
 import (
 	"bytes"
 	"context"
+	"crypto/hmac"
+	"crypto/rand"
+	"crypto/sha256"
 	"io"
 	"io/fs"
+	"math"
 	"os"
 	"path"
 	"path/filepath"
@@ -18,6 +22,7 @@ import (
 	"github.com/efficientgo/core/logerrcapture"
 	"github.com/go-kit/log"
 	"github.com/go-kit/log/level"
+	"github.com/minio/sio"
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
@@ -395,6 +400,103 @@ func DownloadDir(ctx context.Context, logger log.Logger, bkt BucketReader, origi
 // IsOpFailureExpectedFunc allows to mark certain errors as expected, so they will not increment thanos_objstore_bucket_operation_failures_total metric.
 type IsOpFailureExpectedFunc func(error) bool
 
+// BucketWithEncryption takes a bucket and transparently encrypts and decrypts its payloads.
+func BucketWithEncryption(b Bucket, key []byte) *encryptedBucket {
+	return &encryptedBucket{Bucket: b, masterKey: key}
+}
+
+type encryptedBucket struct {
+	Bucket
+
+	masterKey []byte
+}
+
+const saltSizeBytes = 32
+
+// As per https://github.com/minio/sio/blob/master/DARE.md#appendices we need a unique key data stream.
+// We derive a unique key from the configuration provided master key by fetching 32 random bits salt and
+// using KDF(master key ++ salt) as our derived encryption key. The salt is prepended to the encrypted
+// object. This is okay since the salt does not need to be kept a secret.
+func (eb *encryptedBucket) encryptionConfig(salt []byte) sio.Config {
+	h := hmac.New(sha256.New, salt)
+	h.Write(eb.masterKey)
+	key := h.Sum(nil)
+	return sio.Config{Key: key[:], CipherSuites: []byte{sio.AES_256_GCM}}
+}
+
+func (eb *encryptedBucket) Attributes(ctx context.Context, name string) (ObjectAttributes, error) {
+	attrs, err := eb.Bucket.Attributes(ctx, name)
+	if err != nil {
+		return attrs, err
+	}
+
+	decSize, err := sio.DecryptedSize(uint64(attrs.Size) - saltSizeBytes)
+	if err != nil {
+		return ObjectAttributes{}, errors.Wrap(err, "unable to determine unecrypted size")
+	}
+
+	if decSize > math.MaxInt64 {
+		return ObjectAttributes{}, errors.New("size of decrypted blob too large")
+	}
+
+	return ObjectAttributes{Size: int64(decSize), LastModified: attrs.LastModified}, nil
+}
+
+func (eb *encryptedBucket) Upload(ctx context.Context, name string, r io.Reader) error {
+	salt := make([]byte, saltSizeBytes)
+	if _, err := rand.Read(salt); err != nil {
+		return errors.Wrap(err, "unable to derive encryption key for stream")
+	}
+
+	er, err := sio.EncryptReader(r, eb.encryptionConfig(salt))
+	if err != nil {
+		return errors.Wrap(err, "unable to create encryption stream")
+	}
+
+	tr := io.MultiReader(bytes.NewReader(salt), er)
+	return eb.Bucket.Upload(ctx, name, tr)
+}
+
+func (eb *encryptedBucket) Get(ctx context.Context, name string) (io.ReadCloser, error) {
+	return eb.GetRange(ctx, name, 0, -1)
+}
+
+func (eb *encryptedBucket) GetRange(ctx context.Context, name string, off, length int64) (io.ReadCloser, error) {
+	saltReader, err := eb.Bucket.GetRange(ctx, name, 0, saltSizeBytes)
+	if err != nil {
+		return nil, errors.Wrap(err, "unable to fetch salt")
+	}
+	defer saltReader.Close()
+
+	salt, err := io.ReadAll(saltReader)
+	if err != nil {
+		return nil, errors.Wrap(err, "unable to read salt")
+	}
+
+	br := &bucketReaderAt{ctx: ctx, name: name, b: eb.Bucket}
+	dr, err := sio.DecryptReaderAt(br, eb.encryptionConfig(salt))
+	if err != nil {
+		return nil, errors.Wrap(err, "unable to create decryption stream")
+	}
+	return io.NopCloser(io.NewSectionReader(dr, off, length)), nil
+}
+
+type bucketReaderAt struct {
+	ctx  context.Context
+	name string
+	b    BucketReader
+}
+
+func (br *bucketReaderAt) ReadAt(p []byte, off int64) (n int, err error) {
+	rc, err := br.b.GetRange(br.ctx, br.name, off+saltSizeBytes, int64(len(p)))
+	if err != nil {
+		return 0, err
+	}
+	defer rc.Close()
+
+	return rc.Read(p)
+}
+
 var _ InstrumentedBucket = &metricBucket{}
 
 // BucketWithMetrics takes a bucket and registers metrics with the given registry for