feat: implement encryptedStorage

cmd/push.go

@@ -77,7 +77,7 @@ func doPush(opts *pushOptions, cmd *cobra.Command, args []string) {
 		pr, pw := io.Pipe()
 		go func(r io.Reader) {
 			// discard compression header
-			err := c.Compress(util.DiscardN(pw, compressionHeaderSize), r)
+			err := c.Compress(util.DiscardNWriter(pw, compressionHeaderSize), r)
 			pw.CloseWithError(err)
 		}(in)
 		in = pr

pkg/app/globalstorage.go

@@ -8,13 +8,17 @@ import (
 	"strings"
 
 	"github.com/apecloud/datasafed/pkg/config"
+	"github.com/apecloud/datasafed/pkg/encryption"
 	"github.com/apecloud/datasafed/pkg/storage"
+	"github.com/apecloud/datasafed/pkg/storage/encrypted"
 	"github.com/apecloud/datasafed/pkg/storage/kopia"
 	"github.com/apecloud/datasafed/pkg/storage/rclone"
 )
 
 const (
 	backendBasePathEnv   = "DATASAFED_BACKEND_BASE_PATH"
+	encryptionAlgorithm  = "DATASAFED_ENCRYPTION_ALGORITHM"
+	encryptionPassPhrase = "DATASAFED_ENCRYPTION_PASS_PHRASE"
 	kopiaRepoRootEnv     = "DATASAFED_KOPIA_REPO_ROOT"
 	kopiaPasswordEnv     = "DATASAFED_KOPIA_PASSWORD"
 	kopiaDisableCacheEnv = "DATASAFED_KOPIA_DISABLE_CACHE"
@@ -36,15 +40,37 @@ func InitGlobalStorage(ctx context.Context, configFile string) error {
 	storageConf := config.GetGlobal().GetAll(config.StorageSection)
 
 	if kopiaRoot := strings.TrimSpace(os.Getenv(kopiaRepoRootEnv)); kopiaRoot != "" {
-		return initKopiaStorage(ctx, storageConf, basePath, kopiaRoot)
+		err := initKopiaStorage(ctx, storageConf, basePath, kopiaRoot)
+		if err != nil {
+			return err
+		}
 	} else {
 		st, err := createStorage(ctx, storageConf, basePath)
 		if err != nil {
 			return err
 		}
 		globalStorage = st
-		return nil
 	}
+
+	// wrap with encryptedStorage
+	encAlgo := os.Getenv(encryptionAlgorithm)
+	if encAlgo != "" {
+		encPass := os.Getenv(encryptionPassPhrase)
+		if encPass == "" {
+			return fmt.Errorf("encryption pass phrase should not be empty")
+		}
+		enc, err := encryption.CreateEncryptor(encAlgo, []byte(encPass))
+		if err != nil {
+			return err
+		}
+		encSt, err := encrypted.New(ctx, enc, globalStorage)
+		if err != nil {
+			return err
+		}
+		globalStorage = encSt
+	}
+
+	return nil
 }
 
 func GetGlobalStorage() (storage.Storage, error) {

pkg/encryption/aes_encryptor.go

@@ -0,0 +1,96 @@
+package encryption
+
+import (
+	"crypto/aes"
+	"crypto/cipher"
+	"crypto/rand"
+	"fmt"
+	"io"
+)
+
+const bufferSize = ((128 * 1024) / aes.BlockSize) * aes.BlockSize
+
+type aesEncryptor struct {
+	key          []byte
+	newEncryptor func(block cipher.Block, iv []byte) cipher.Stream
+	newDecryptor func(block cipher.Block, iv []byte) cipher.Stream
+}
+
+func (e *aesEncryptor) EncryptStream(plainText io.Reader, output io.Writer) error {
+	iv := make([]byte, aes.BlockSize)
+	if _, err := io.ReadFull(rand.Reader, iv); err != nil {
+		return fmt.Errorf("rand.Read() error: %w", err)
+	}
+	n, err := output.Write(iv)
+	if err != nil {
+		return fmt.Errorf("write IV error: %w", err)
+	}
+	if n != len(iv) {
+		return fmt.Errorf("partially write IV, len: %d, written: %d", len(iv), n)
+	}
+	block, err := aes.NewCipher(e.key)
+	if err != nil {
+		return fmt.Errorf("aes.NewCipher() error: %w", err)
+	}
+	enc := e.newEncryptor(block, iv)
+	buf := make([]byte, bufferSize)
+	return pipeStream(plainText, "plainText", output, "cipherText", buf, func(b []byte) {
+		enc.XORKeyStream(b, b)
+	})
+}
+
+func (e *aesEncryptor) DecryptStream(cipherText io.Reader, output io.Writer) error {
+	iv := make([]byte, aes.BlockSize)
+	_, err := io.ReadFull(cipherText, iv)
+	if err != nil {
+		return fmt.Errorf("unable to read iv from cipherText, error: %w", err)
+	}
+	block, err := aes.NewCipher(e.key)
+	if err != nil {
+		return fmt.Errorf("aes.NewCipher() error: %w", err)
+	}
+	dec := e.newDecryptor(block, iv)
+	buf := make([]byte, bufferSize)
+	return pipeStream(cipherText, "cipherText", output, "plainText", buf, func(b []byte) {
+		dec.XORKeyStream(b, b)
+	})
+}
+
+func (e *aesEncryptor) Overhead() int {
+	return aes.BlockSize
+}
+
+func newAESCFB(passPhrase []byte, keyLength int) (StreamEncryptor, error) {
+	deriveLength := keyLength
+	if deriveLength < minDerivedKeyLength {
+		deriveLength = minDerivedKeyLength
+	}
+	key, err := deriveKey(passPhrase, []byte(purposeEncryptionKey), deriveLength)
+	if err != nil {
+		return nil, fmt.Errorf("deriveKey() error: %w", err)
+	}
+	ae := &aesEncryptor{
+		key:          key[:keyLength],
+		newEncryptor: cipher.NewCFBEncrypter,
+		newDecryptor: cipher.NewCFBDecrypter,
+	}
+	return ae, nil
+}
+
+func NewAES128CFB(passPhrase []byte) (StreamEncryptor, error) {
+	return newAESCFB(passPhrase, 16)
+}
+
+func NewAES192CFB(passPhrase []byte) (StreamEncryptor, error) {
+	return newAESCFB(passPhrase, 24)
+}
+
+func NewAES256CFB(passPhrase []byte) (StreamEncryptor, error) {
+	return newAESCFB(passPhrase, 32)
+}
+
+func init() {
+	Register("AES-128-CFB", "AES-128 with CFB mode", NewAES128CFB)
+	Register("AES-192-CFB", "AES-192 with CFB mode", NewAES192CFB)
+	Register("AES-256-CFB", "AES-256 with CFB mode", NewAES256CFB)
+}

pkg/encryption/common.go

@@ -0,0 +1,55 @@
+package encryption
+
+import (
+	"crypto/sha256"
+	"errors"
+	"fmt"
+	"io"
+
+	"golang.org/x/crypto/hkdf"
+)
+
+const minDerivedKeyLength = 32
+
+func pipeStream(in io.Reader, inName string,
+	out io.Writer, outName string,
+	buf []byte, manipulateFunc func([]byte)) error {
+	for {
+		n, err := in.Read(buf)
+		if n > 0 {
+			data := buf[:n]
+			manipulateFunc(data)
+			wn, err := out.Write(data)
+			if err != nil {
+				return fmt.Errorf("write to %s error: %w", outName, err)
+			}
+			if wn != n {
+				return fmt.Errorf("partially write to %s, length: %d, actual write: %d",
+					outName, n, wn)
+			}
+		}
+		if err != nil {
+			if errors.Is(err, io.EOF) {
+				return nil
+			} else {
+				return fmt.Errorf("read from %s error: %w", inName, err)
+			}
+		}
+	}
+}
+
+// deriveKey uses HKDF to derive a key of a given length and a given purpose from parameters.
+func deriveKey(passPhrase []byte, purpose []byte, length int) ([]byte, error) {
+	if length < minDerivedKeyLength {
+		return nil, fmt.Errorf("derived key must be at least 32 bytes, was %v", length)
+	}
+
+	key := make([]byte, length)
+	k := hkdf.New(sha256.New, passPhrase, purpose, nil)
+	_, err := io.ReadFull(k, key)
+	if err != nil {
+		return nil, err
+	}
+
+	return key, nil
+}

pkg/encryption/encryption.go

@@ -0,0 +1,53 @@
+// Package encryption manages content encryption algorithms.
+package encryption
+
+import (
+	"sort"
+	"strings"
+
+	"github.com/pkg/errors"
+)
+
+const (
+	purposeEncryptionKey = "encryption"
+)
+
+// CreateEncryptor creates an StreamEncryptor for given parameters.
+func CreateEncryptor(algorithm string, passPhrase []byte) (StreamEncryptor, error) {
+	algorithm = strings.ToUpper(algorithm)
+	e := encryptors[algorithm]
+	if e == nil {
+		return nil, errors.Errorf("unknown encryption algorithm: %v", algorithm)
+	}
+
+	return e.newEncryptor(passPhrase)
+}
+
+// EncryptorFactory creates new Encryptor for given parameters.
+type EncryptorFactory func(passPhrase []byte) (StreamEncryptor, error)
+
+// SupportedAlgorithms returns the names of the supported encryption methods.
+func SupportedAlgorithms() []string {
+	var result []string
+	for k := range encryptors {
+		result = append(result, k)
+	}
+	sort.Strings(result)
+	return result
+}
+
+// Register registers new encryption algorithm.
+func Register(name, description string, newEncryptor EncryptorFactory) {
+	name = strings.ToUpper(name)
+	encryptors[name] = &encryptorInfo{
+		description,
+		newEncryptor,
+	}
+}
+
+type encryptorInfo struct {
+	description  string
+	newEncryptor EncryptorFactory
+}
+
+var encryptors = map[string]*encryptorInfo{}