Merge branch 'master' into mq-subscribe

5 files changed, 43 insertions, 2 deletions

diff --git a/weed/server/filer_server.go b/weed/server/filer_server.go
index 356761f30..9880afee0 100644
--- a/weed/server/filer_server.go
+++ b/weed/server/filer_server.go
@@ -91,6 +91,7 @@ type FilerServer struct {
 	secret         security.SigningKey
 	filer          *filer.Filer
 	filerGuard     *security.Guard
+	volumeGuard    *security.Guard
 	grpcDialOption grpc.DialOption
 
 	// metrics read from the master
@@ -113,6 +114,14 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption)
 	v.SetDefault("jwt.filer_signing.read.expires_after_seconds", 60)
 	readExpiresAfterSec := v.GetInt("jwt.filer_signing.read.expires_after_seconds")
 
+	volumeSigningKey := v.GetString("jwt.signing.key")
+	v.SetDefault("jwt.signing.expires_after_seconds", 10)
+	volumeExpiresAfterSec := v.GetInt("jwt.signing.expires_after_seconds")
+
+	volumeReadSigningKey := v.GetString("jwt.signing.read.key")
+	v.SetDefault("jwt.signing.read.expires_after_seconds", 60)
+	volumeReadExpiresAfterSec := v.GetInt("jwt.signing.read.expires_after_seconds")
+
 	v.SetDefault("cors.allowed_origins.values", "*")
 
 	allowedOrigins := v.GetString("cors.allowed_origins.values")
@@ -145,6 +154,7 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption)
 	fs.filer.Cipher = option.Cipher
 	// we do not support IP whitelist right now
 	fs.filerGuard = security.NewGuard([]string{}, signingKey, expiresAfterSec, readSigningKey, readExpiresAfterSec)
+	fs.volumeGuard = security.NewGuard([]string{}, volumeSigningKey, volumeExpiresAfterSec, volumeReadSigningKey, volumeReadExpiresAfterSec)
 
 	fs.checkWithMaster()
 
diff --git a/weed/server/filer_server_handlers_proxy.go b/weed/server/filer_server_handlers_proxy.go
index db46f00b3..e04994569 100644
--- a/weed/server/filer_server_handlers_proxy.go
+++ b/weed/server/filer_server_handlers_proxy.go
@@ -2,6 +2,7 @@ package weed_server
 
 import (
 	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/security"
 	"github.com/seaweedfs/seaweedfs/weed/util"
 	"github.com/seaweedfs/seaweedfs/weed/util/mem"
 	"io"
@@ -20,6 +21,26 @@ func init() {
 	}}
 }
 
+func (fs *FilerServer) maybeAddVolumeJwtAuthorization(r *http.Request, fileId string, isWrite bool) {
+	encodedJwt := fs.maybeGetVolumeJwtAuthorizationToken(fileId, isWrite)
+
+	if encodedJwt == "" {
+		return
+	}
+
+	r.Header.Set("Authorization", "BEARER "+string(encodedJwt))
+}
+
+func (fs *FilerServer) maybeGetVolumeJwtAuthorizationToken(fileId string, isWrite bool) string {
+	var encodedJwt security.EncodedJwt
+	if isWrite {
+		encodedJwt = security.GenJwtForVolumeServer(fs.volumeGuard.SigningKey, fs.volumeGuard.ExpiresAfterSec, fileId)
+	} else {
+		encodedJwt = security.GenJwtForVolumeServer(fs.volumeGuard.ReadSigningKey, fs.volumeGuard.ReadExpiresAfterSec, fileId)
+	}
+	return string(encodedJwt)
+}
+
 func (fs *FilerServer) proxyToVolumeServer(w http.ResponseWriter, r *http.Request, fileId string) {
 
 	urlStrings, err := fs.filer.MasterClient.GetLookupFileIdFunction()(fileId)
diff --git a/weed/server/filer_server_handlers_read.go b/weed/server/filer_server_handlers_read.go
index c139060e4..83411051d 100644
--- a/weed/server/filer_server_handlers_read.go
+++ b/weed/server/filer_server_handlers_read.go
@@ -15,6 +15,7 @@ import (
 	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/seaweedfs/seaweedfs/weed/security"
 	"github.com/seaweedfs/seaweedfs/weed/util/mem"
 
 	"github.com/seaweedfs/seaweedfs/weed/filer"
@@ -261,7 +262,7 @@ func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request)
 			}
 		}
 
-		streamFn, err := filer.PrepareStreamContentWithThrottler(fs.filer.MasterClient, chunks, offset, size, fs.option.DownloadMaxBytesPs)
+		streamFn, err := filer.PrepareStreamContentWithThrottler(fs.filer.MasterClient, fs.maybeGetVolumeReadJwtAuthorizationToken, chunks, offset, size, fs.option.DownloadMaxBytesPs)
 		if err != nil {
 			stats.FilerHandlerCounter.WithLabelValues(stats.ErrorReadStream).Inc()
 			glog.Errorf("failed to prepare stream content %s: %v", r.URL, err)
@@ -277,3 +278,7 @@ func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request)
 		}, nil
 	})
 }
+
+func (fs *FilerServer) maybeGetVolumeReadJwtAuthorizationToken(fileId string) string {
+	return string(security.GenJwtForVolumeServer(fs.volumeGuard.ReadSigningKey, fs.volumeGuard.ReadExpiresAfterSec, fileId))
+}
diff --git a/weed/server/master_grpc_server_assign.go b/weed/server/master_grpc_server_assign.go
index 2aede2d50..5839a6a73 100644
--- a/weed/server/master_grpc_server_assign.go
+++ b/weed/server/master_grpc_server_assign.go
@@ -97,6 +97,9 @@ func (ms *MasterServer) Assign(ctx context.Context, req *master_pb.AssignRequest
 			continue
 		}
 		dn := dnList.Head()
+		if dn == nil {
+			continue
+		}
 		var replicas []*master_pb.Location
 		for _, r := range dnList.Rest() {
 			replicas = append(replicas, &master_pb.Location{
diff --git a/weed/server/master_server_handlers.go b/weed/server/master_server_handlers.go
index c5e059f21..9dc6351a4 100644
--- a/weed/server/master_server_handlers.go
+++ b/weed/server/master_server_handlers.go
@@ -149,7 +149,9 @@ func (ms *MasterServer) dirAssignHandler(w http.ResponseWriter, r *http.Request)
 		} else {
 			ms.maybeAddJwtAuthorization(w, fid, true)
 			dn := dnList.Head()
-
+			if dn == nil {
+				continue
+			}
 			writeJsonQuiet(w, r, http.StatusOK, operation.AssignResult{Fid: fid, Url: dn.Url(), PublicUrl: dn.PublicUrl, Count: count})
 			return
 		}