filer: option to encrypt data on volume server

6 files changed, 62 insertions, 27 deletions

diff --git a/weed/command/filer.go b/weed/command/filer.go
index b5b595215..327ee8316 100644
--- a/weed/command/filer.go
+++ b/weed/command/filer.go
@@ -34,6 +34,7 @@ type FilerOptions struct {
 	dataCenter              *string
 	enableNotification      *bool
 	disableHttp             *bool
+	cipher                  *bool
 
 	// default leveldb directory, used in "weed server" mode
 	defaultLevelDbDirectory *string
@@ -53,6 +54,7 @@ func init() {
 	f.dirListingLimit = cmdFiler.Flag.Int("dirListLimit", 100000, "limit sub dir listing size")
 	f.dataCenter = cmdFiler.Flag.String("dataCenter", "", "prefer to write to volumes in this data center")
 	f.disableHttp = cmdFiler.Flag.Bool("disableHttp", false, "disable http request, only gRpc operations are allowed")
+	f.cipher = cmdFiler.Flag.Bool("encryptVolumeData", false, "encrypt data on volume servers")
 }
 
 var cmdFiler = &Command{
@@ -111,6 +113,7 @@ func (fo *FilerOptions) startFiler() {
 		DefaultLevelDbDir:  defaultLevelDbDirectory,
 		DisableHttp:        *fo.disableHttp,
 		Port:               uint32(*fo.port),
+		Cipher:             *fo.cipher,
 	})
 	if nfs_err != nil {
 		glog.Fatalf("Filer startup error: %v", nfs_err)
diff --git a/weed/command/filer_copy.go b/weed/command/filer_copy.go
index 3e7ae1db2..8c01cfd74 100644
--- a/weed/command/filer_copy.go
+++ b/weed/command/filer_copy.go
@@ -41,6 +41,7 @@ type CopyOptions struct {
 	compressionLevel  *int
 	grpcDialOption    grpc.DialOption
 	masters           []string
+	cipher            bool
 }
 
 func init() {
@@ -108,7 +109,7 @@ func runCopy(cmd *Command, args []string) bool {
 	filerGrpcAddress := fmt.Sprintf("%s:%d", filerUrl.Hostname(), filerGrpcPort)
 	copy.grpcDialOption = security.LoadClientTLS(util.GetViper(), "grpc.client")
 
-	masters, collection, replication, maxMB, err := readFilerConfiguration(copy.grpcDialOption, filerGrpcAddress)
+	masters, collection, replication, maxMB, cipher, err := readFilerConfiguration(copy.grpcDialOption, filerGrpcAddress)
 	if err != nil {
 		fmt.Printf("read from filer %s: %v\n", filerGrpcAddress, err)
 		return false
@@ -123,6 +124,7 @@ func runCopy(cmd *Command, args []string) bool {
 		*copy.maxMB = int(maxMB)
 	}
 	copy.masters = masters
+	copy.cipher = cipher
 
 	if *cmdCopy.IsDebug {
 		util.SetupProfiling("filer.copy.cpu.pprof", "filer.copy.mem.pprof")
@@ -159,13 +161,14 @@ func runCopy(cmd *Command, args []string) bool {
 	return true
 }
 
-func readFilerConfiguration(grpcDialOption grpc.DialOption, filerGrpcAddress string) (masters []string, collection, replication string, maxMB uint32, err error) {
+func readFilerConfiguration(grpcDialOption grpc.DialOption, filerGrpcAddress string) (masters []string, collection, replication string, maxMB uint32, cipher bool, err error) {
 	err = pb.WithGrpcFilerClient(filerGrpcAddress, grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
 		resp, err := client.GetFilerConfiguration(context.Background(), &filer_pb.GetFilerConfigurationRequest{})
 		if err != nil {
 			return fmt.Errorf("get filer %s configuration: %v", filerGrpcAddress, err)
 		}
 		masters, collection, replication, maxMB = resp.Masters, resp.Collection, resp.Replication, resp.MaxMb
+		cipher = resp.Cipher
 		return nil
 	})
 	return
@@ -300,7 +303,7 @@ func (worker *FileCopyWorker) uploadFileAsOne(task FileCopyTask, f *os.File) err
 
 		targetUrl := "http://" + assignResult.Url + "/" + assignResult.FileId
 
-		uploadResult, err := operation.UploadWithLocalCompressionLevel(targetUrl, fileName, f, false, mimeType, nil, security.EncodedJwt(assignResult.Auth), *worker.options.compressionLevel)
+		uploadResult, err := operation.UploadWithLocalCompressionLevel(targetUrl, fileName, worker.options.cipher, f, false, mimeType, nil, security.EncodedJwt(assignResult.Auth), *worker.options.compressionLevel)
 		if err != nil {
 			return fmt.Errorf("upload data %v to %s: %v\n", fileName, targetUrl, err)
 		}
@@ -310,11 +313,12 @@ func (worker *FileCopyWorker) uploadFileAsOne(task FileCopyTask, f *os.File) err
 		fmt.Printf("uploaded %s to %s\n", fileName, targetUrl)
 
 		chunks = append(chunks, &filer_pb.FileChunk{
-			FileId: assignResult.FileId,
-			Offset: 0,
-			Size:   uint64(uploadResult.Size),
-			Mtime:  time.Now().UnixNano(),
-			ETag:   uploadResult.ETag,
+			FileId:    assignResult.FileId,
+			Offset:    0,
+			Size:      uint64(uploadResult.Size),
+			Mtime:     time.Now().UnixNano(),
+			ETag:      uploadResult.ETag,
+			CipherKey: uploadResult.CipherKey,
 		})
 
 		fmt.Printf("copied %s => http://%s%s%s\n", fileName, worker.filerHost, task.destinationUrlPath, fileName)
@@ -409,10 +413,7 @@ func (worker *FileCopyWorker) uploadFileInChunks(task FileCopyTask, f *os.File,
 				replication = assignResult.Replication
 			}
 
-			uploadResult, err := operation.Upload(targetUrl,
-				fileName+"-"+strconv.FormatInt(i+1, 10),
-				io.NewSectionReader(f, i*chunkSize, chunkSize),
-				false, "", nil, security.EncodedJwt(assignResult.Auth))
+			uploadResult, err := operation.Upload(targetUrl, fileName+"-"+strconv.FormatInt(i+1, 10), false, io.NewSectionReader(f, i*chunkSize, chunkSize), false, "", nil, security.EncodedJwt(assignResult.Auth))
 			if err != nil {
 				uploadError = fmt.Errorf("upload data %v to %s: %v\n", fileName, targetUrl, err)
 				return
@@ -422,11 +423,12 @@ func (worker *FileCopyWorker) uploadFileInChunks(task FileCopyTask, f *os.File,
 				return
 			}
 			chunksChan <- &filer_pb.FileChunk{
-				FileId: assignResult.FileId,
-				Offset: i * chunkSize,
-				Size:   uint64(uploadResult.Size),
-				Mtime:  time.Now().UnixNano(),
-				ETag:   uploadResult.ETag,
+				FileId:    assignResult.FileId,
+				Offset:    i * chunkSize,
+				Size:      uint64(uploadResult.Size),
+				Mtime:     time.Now().UnixNano(),
+				ETag:      uploadResult.ETag,
+				CipherKey: uploadResult.CipherKey,
 			}
 			fmt.Printf("uploaded %s-%d to %s [%d,%d)\n", fileName, i+1, targetUrl, i*chunkSize, i*chunkSize+int64(uploadResult.Size))
 		}(i)
diff --git a/weed/command/mount_std.go b/weed/command/mount_std.go
index b195bf143..9177091a5 100644
--- a/weed/command/mount_std.go
+++ b/weed/command/mount_std.go
@@ -145,11 +145,13 @@ func RunMount(filer, filerMountRootPath, dir, collection, replication, dataCente
 
 	// try to connect to filer, filerBucketsPath may be useful later
 	grpcDialOption := security.LoadClientTLS(util.GetViper(), "grpc.client")
+	var cipher bool
 	err = pb.WithGrpcFilerClient(filerGrpcAddress, grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
-		_, err := client.GetFilerConfiguration(context.Background(), &filer_pb.GetFilerConfigurationRequest{})
+		resp, err := client.GetFilerConfiguration(context.Background(), &filer_pb.GetFilerConfigurationRequest{})
 		if err != nil {
 			return fmt.Errorf("get filer %s configuration: %v", filerGrpcAddress, err)
 		}
+		cipher = resp.Cipher
 		return nil
 	})
 	if err != nil {
@@ -183,6 +185,7 @@ func RunMount(filer, filerMountRootPath, dir, collection, replication, dataCente
 		MountMtime:                  time.Now(),
 		Umask:                       umask,
 		OutsideContainerClusterMode: outsideContainerClusterMode,
+		Cipher:                      cipher,
 	}))
 	if err != nil {
 		fuse.Unmount(dir)
diff --git a/weed/command/scaffold.go b/weed/command/scaffold.go
index 5b246b7c0..f4a08fb51 100644
--- a/weed/command/scaffold.go
+++ b/weed/command/scaffold.go
@@ -75,9 +75,9 @@ const (
 # recursive_delete will delete all sub folders and files, similar to "rm -Rf"
 recursive_delete = false
 # directories under this folder will be automatically creating a separate bucket
-buckets_folder = /buckets
+buckets_folder = "/buckets"
 # directories under this folder will be store message queue data
-queues_folder = /queues
+queues_folder = "/queues"
 
 ####################################################
 # The following are filer store options
diff --git a/weed/command/server.go b/weed/command/server.go
index c9d27555c..f45429193 100644
--- a/weed/command/server.go
+++ b/weed/command/server.go
@@ -82,6 +82,7 @@ func init() {
 	filerOptions.disableDirListing = cmdServer.Flag.Bool("filer.disableDirListing", false, "turn off directory listing")
 	filerOptions.maxMB = cmdServer.Flag.Int("filer.maxMB", 32, "split files larger than the limit")
 	filerOptions.dirListingLimit = cmdServer.Flag.Int("filer.dirListLimit", 1000, "limit sub dir listing size")
+	filerOptions.cipher = cmdServer.Flag.Bool("filer.encryptVolumeData", false, "encrypt data on volume servers")
 
 	serverOptions.v.port = cmdServer.Flag.Int("volume.port", 8080, "volume server http listen port")
 	serverOptions.v.publicPort = cmdServer.Flag.Int("volume.port.public", 0, "volume server public port")
diff --git a/weed/command/webdav.go b/weed/command/webdav.go
index ba88a17be..4f5d5f5ce 100644
--- a/weed/command/webdav.go
+++ b/weed/command/webdav.go
@@ -1,6 +1,7 @@
 package command
 
 import (
+	"context"
 	"fmt"
 	"net/http"
 	"os/user"
@@ -9,6 +10,7 @@ import (
 
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/pb"
+	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
 	"github.com/chrislusf/seaweedfs/weed/security"
 	"github.com/chrislusf/seaweedfs/weed/server"
 	"github.com/chrislusf/seaweedfs/weed/util"
@@ -55,12 +57,6 @@ func runWebDav(cmd *Command, args []string) bool {
 
 func (wo *WebDavOption) startWebDav() bool {
 
-	filerGrpcAddress, err := pb.ParseFilerGrpcAddress(*wo.filer)
-	if err != nil {
-		glog.Fatal(err)
-		return false
-	}
-
 	// detect current user
 	uid, gid := uint32(0), uint32(0)
 	if u, err := user.Current(); err == nil {
@@ -72,13 +68,43 @@ func (wo *WebDavOption) startWebDav() bool {
 		}
 	}
 
+	// parse filer grpc address
+	filerGrpcAddress, err := pb.ParseFilerGrpcAddress(*wo.filer)
+	if err != nil {
+		glog.Fatal(err)
+		return false
+	}
+
+	grpcDialOption := security.LoadClientTLS(util.GetViper(), "grpc.client")
+
+	var cipher bool
+	// connect to filer
+	for {
+		err = pb.WithGrpcFilerClient(filerGrpcAddress, grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
+			resp, err := client.GetFilerConfiguration(context.Background(), &filer_pb.GetFilerConfigurationRequest{})
+			if err != nil {
+				return fmt.Errorf("get filer %s configuration: %v", filerGrpcAddress, err)
+			}
+			cipher = resp.Cipher
+			return nil
+		})
+		if err != nil {
+			glog.V(0).Infof("wait to connect to filer %s grpc address %s", *wo.filer, filerGrpcAddress)
+			time.Sleep(time.Second)
+		} else {
+			glog.V(0).Infof("connected to filer %s grpc address %s", *wo.filer, filerGrpcAddress)
+			break
+		}
+	}
+
 	ws, webdavServer_err := weed_server.NewWebDavServer(&weed_server.WebDavOption{
 		Filer:            *wo.filer,
 		FilerGrpcAddress: filerGrpcAddress,
-		GrpcDialOption:   security.LoadClientTLS(util.GetViper(), "grpc.client"),
+		GrpcDialOption:   grpcDialOption,
 		Collection:       *wo.collection,
 		Uid:              uid,
 		Gid:              gid,
+		Cipher:           cipher,
 	})
 	if webdavServer_err != nil {
 		glog.Fatalf("WebDav Server startup error: %v", webdavServer_err)