Merge branch 'master' into mq

1 files changed, 86 insertions, 16 deletions

diff --git a/weed/command/filer.go b/weed/command/filer.go
index 877c4b5d5..b7f67ea3b 100644
--- a/weed/command/filer.go
+++ b/weed/command/filer.go
@@ -1,6 +1,9 @@
 package command
 
 import (
+	"context"
+	"crypto/tls"
+	"crypto/x509"
 	"fmt"
 	"net"
 	"net/http"
@@ -10,8 +13,6 @@ import (
 	"strings"
 	"time"
 
-	"google.golang.org/grpc/reflection"
-
 	"github.com/seaweedfs/seaweedfs/weed/filer"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb"
@@ -20,6 +21,10 @@ import (
 	weed_server "github.com/seaweedfs/seaweedfs/weed/server"
 	stats_collect "github.com/seaweedfs/seaweedfs/weed/stats"
 	"github.com/seaweedfs/seaweedfs/weed/util"
+	"github.com/spf13/viper"
+	"google.golang.org/grpc/credentials/tls/certprovider"
+	"google.golang.org/grpc/credentials/tls/certprovider/pemfile"
+	"google.golang.org/grpc/reflection"
 )
 
 var (
@@ -52,6 +57,7 @@ type FilerOptions struct {
 	disableHttp             *bool
 	cipher                  *bool
 	metricsHttpPort         *int
+	metricsHttpIp           *string
 	saveToFilerLimit        *int
 	defaultLevelDbDirectory *string
 	concurrentUploadLimitMB *int
@@ -63,7 +69,7 @@ type FilerOptions struct {
 	diskType                *string
 	allowedOrigins          *string
 	exposeDirectoryData     *bool
-	joinExistingFiler       *bool
+	certProvider            certprovider.Provider
 }
 
 func init() {
@@ -85,6 +91,7 @@ func init() {
 	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")
 	f.metricsHttpPort = cmdFiler.Flag.Int("metricsPort", 0, "Prometheus metrics listen port")
+	f.metricsHttpIp = cmdFiler.Flag.String("metricsIp", "", "metrics listen ip. If empty, default to same as -ip.bind option.")
 	f.saveToFilerLimit = cmdFiler.Flag.Int("saveToFilerLimit", 0, "files smaller than this limit will be saved in filer store")
 	f.defaultLevelDbDirectory = cmdFiler.Flag.String("defaultStoreDir", ".", "if filer.toml is empty, use an embedded filer store in the directory")
 	f.concurrentUploadLimitMB = cmdFiler.Flag.Int("concurrentUploadLimitMB", 128, "limit total concurrent upload size")
@@ -96,7 +103,6 @@ func init() {
 	f.diskType = cmdFiler.Flag.String("disk", "", "[hdd|ssd|<tag>] hard drive or solid state drive or any tag")
 	f.allowedOrigins = cmdFiler.Flag.String("allowedOrigins", "*", "comma separated list of allowed origins")
 	f.exposeDirectoryData = cmdFiler.Flag.Bool("exposeDirectoryData", true, "whether to return directory metadata and content in Filer UI")
-	f.joinExistingFiler = cmdFiler.Flag.Bool("joinExistingFiler", false, "enable if new filer wants to join existing cluster")
 
 	// start s3 on filer
 	filerStartS3 = cmdFiler.Flag.Bool("s3", false, "whether to start S3 gateway")
@@ -124,6 +130,7 @@ func init() {
 	filerWebDavOptions.tlsCertificate = cmdFiler.Flag.String("webdav.cert.file", "", "path to the TLS certificate file")
 	filerWebDavOptions.cacheDir = cmdFiler.Flag.String("webdav.cacheDir", os.TempDir(), "local cache directory for file chunks")
 	filerWebDavOptions.cacheSizeMB = cmdFiler.Flag.Int64("webdav.cacheCapacityMB", 0, "local cache capacity in MB")
+	filerWebDavOptions.maxMB = cmdFiler.Flag.Int("webdav.maxMB", 4, "split files larger than the limit")
 	filerWebDavOptions.filerRootPath = cmdFiler.Flag.String("webdav.filer.path", "/", "use this remote path from filer server")
 
 	// start iam on filer
@@ -172,9 +179,17 @@ func runFiler(cmd *Command, args []string) bool {
 		go http.ListenAndServe(fmt.Sprintf(":%d", *f.debugPort), nil)
 	}
 
-	util.LoadConfiguration("security", false)
+	util.LoadSecurityConfiguration()
 
-	go stats_collect.StartMetricsServer(*f.bindIp, *f.metricsHttpPort)
+	switch {
+	case *f.metricsHttpIp != "":
+		// noting to do, use f.metricsHttpIp
+	case *f.bindIp != "":
+		*f.metricsHttpIp = *f.bindIp
+	case *f.ip != "":
+		*f.metricsHttpIp = *f.ip
+	}
+	go stats_collect.StartMetricsServer(*f.metricsHttpIp, *f.metricsHttpPort)
 
 	filerAddress := pb.NewServerAddress(*f.ip, *f.port, *f.portGrpc).String()
 	startDelay := time.Duration(2)
@@ -222,6 +237,15 @@ func runFiler(cmd *Command, args []string) bool {
 	return true
 }
 
+// GetCertificateWithUpdate Auto refreshing TSL certificate
+func (fo *FilerOptions) GetCertificateWithUpdate(*tls.ClientHelloInfo) (*tls.Certificate, error) {
+	certs, err := fo.certProvider.KeyMaterial(context.Background())
+	if certs == nil {
+		return nil, err
+	}
+	return &certs.Certs[0], err
+}
+
 func (fo *FilerOptions) startFiler() {
 
 	defaultMux := http.NewServeMux()
@@ -264,7 +288,6 @@ func (fo *FilerOptions) startFiler() {
 		DownloadMaxBytesPs:    int64(*fo.downloadMaxMBps) * 1024 * 1024,
 		DiskType:              *fo.diskType,
 		AllowedOrigins:        strings.Split(*fo.allowedOrigins, ","),
-		JoinExistingFiler:     *fo.joinExistingFiler,
 	})
 	if nfs_err != nil {
 		glog.Fatalf("Filer startup error: %v", nfs_err)
@@ -332,15 +355,62 @@ func (fo *FilerOptions) startFiler() {
 			httpS.Serve(filerSocketListener)
 		}()
 	}
-	if filerLocalListener != nil {
-		go func() {
-			if err := httpS.Serve(filerLocalListener); err != nil {
-				glog.Errorf("Filer Fail to serve: %v", e)
+
+	if viper.GetString("https.filer.key") != "" {
+		certFile := viper.GetString("https.filer.cert")
+		keyFile := viper.GetString("https.filer.key")
+		caCertFile := viper.GetString("https.filer.ca")
+		disbaleTlsVerifyClientCert := viper.GetBool("https.filer.disable_tls_verify_client_cert")
+
+		pemfileOptions := pemfile.Options{
+			CertFile:        certFile,
+			KeyFile:         keyFile,
+			RefreshDuration: security.CredRefreshingInterval,
+		}
+		if fo.certProvider, err = pemfile.NewProvider(pemfileOptions); err != nil {
+			glog.Fatalf("pemfile.NewProvider(%v) failed: %v", pemfileOptions, err)
+		}
+
+		caCertPool := x509.NewCertPool()
+		if caCertFile != "" {
+			caCertFile, err := os.ReadFile(caCertFile)
+			if err != nil {
+				glog.Fatalf("error reading CA certificate: %v", err)
 			}
-		}()
-	}
-	if err := httpS.Serve(filerListener); err != nil {
-		glog.Fatalf("Filer Fail to serve: %v", e)
-	}
+			caCertPool.AppendCertsFromPEM(caCertFile)
+		}
+
+		clientAuth := tls.NoClientCert
+		if !disbaleTlsVerifyClientCert {
+			clientAuth = tls.RequireAndVerifyClientCert
+		}
+
+		httpS.TLSConfig = &tls.Config{
+			GetCertificate: fo.GetCertificateWithUpdate,
+			ClientAuth:     clientAuth,
+			ClientCAs:      caCertPool,
+		}
 
+		if filerLocalListener != nil {
+			go func() {
+				if err := httpS.ServeTLS(filerLocalListener, "", ""); err != nil {
+					glog.Errorf("Filer Fail to serve: %v", e)
+				}
+			}()
+		}
+		if err := httpS.ServeTLS(filerListener, "", ""); err != nil {
+			glog.Fatalf("Filer Fail to serve: %v", e)
+		}
+	} else {
+		if filerLocalListener != nil {
+			go func() {
+				if err := httpS.Serve(filerLocalListener); err != nil {
+					glog.Errorf("Filer Fail to serve: %v", e)
+				}
+			}()
+		}
+		if err := httpS.Serve(filerListener); err != nil {
+			glog.Fatalf("Filer Fail to serve: %v", e)
+		}
+	}
 }