gsh-digital-services/seaweedFS
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

filer cipher: single chunk http POST and PUT and read

Browse Source
This commit is contained in:
Chris Lu
2020-03-07 06:06:58 -08:00
parent e3b8bf5588
commit ea1169dc80
10 changed files with 354 additions and 212 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
weed/server/common.go
View File

@@ -99,13 +99,13 @@ func submitForClientHandler(w http.ResponseWriter, r *http.Request, masterUrl st
}
debug("parsing upload file...")
fname, data, mimeType, pairMap, isGzipped, originalDataSize, lastModified, _, _, pe := needle.ParseUpload(r, 256*1024*1024)
pu, pe := needle.ParseUpload(r, 256*1024*1024)
if pe != nil {
writeJsonError(w, r, http.StatusBadRequest, pe)
return
}
debug("assigning file id for", fname)
debug("assigning file id for", pu.FileName)
r.ParseForm()
count := uint64(1)
if r.FormValue("count") != "" {
@@ -129,21 +129,21 @@ func submitForClientHandler(w http.ResponseWriter, r *http.Request, masterUrl st
}
url := "http://" + assignResult.Url + "/" + assignResult.Fid
if lastModified != 0 {
url = url + "?ts=" + strconv.FormatUint(lastModified, 10)
if pu.ModifiedTime != 0 {
url = url + "?ts=" + strconv.FormatUint(pu.ModifiedTime, 10)
}
debug("upload file to store", url)
uploadResult, err := operation.Upload(url, fname, false, bytes.NewReader(data), isGzipped, mimeType, pairMap, assignResult.Auth)
uploadResult, err := operation.Upload(url, pu.FileName, false, bytes.NewReader(pu.Data), pu.IsGzipped, pu.MimeType, pu.PairMap, assignResult.Auth)
if err != nil {
writeJsonError(w, r, http.StatusInternalServerError, err)
return
}
m["fileName"] = fname
m["fileName"] = pu.FileName
m["fid"] = assignResult.Fid
m["fileUrl"] = assignResult.PublicUrl + "/" + assignResult.Fid
m["size"] = originalDataSize
m["size"] = pu.OriginalDataSize
m["eTag"] = uploadResult.ETag
writeJsonQuiet(w, r, http.StatusCreated, m)
return

11
weed/server/filer_server_handlers_read.go
View File

@@ -2,6 +2,7 @@ package weed_server
import (
"context"
"fmt"
"io"
"io/ioutil"
"mime"
@@ -14,7 +15,6 @@ import (
"github.com/chrislusf/seaweedfs/weed/filer2"
"github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
"github.com/chrislusf/seaweedfs/weed/stats"
"github.com/chrislusf/seaweedfs/weed/util"
)
@@ -136,15 +136,16 @@ func (fs *FilerServer) handleSingleChunk(w http.ResponseWriter, r *http.Request,
if entry.Attr.Mime != "" {
w.Header().Set("Content-Type", entry.Attr.Mime)
}
w.WriteHeader(resp.StatusCode)
if entry.Chunks[0].CipherKey == nil {
w.WriteHeader(resp.StatusCode)
io.Copy(w, resp.Body)
} else {
fs.writeEncryptedChunk(w, resp, entry.Chunks[0])
fs.writeEncryptedChunk(w, resp, entry)
}
}
func (fs *FilerServer) writeEncryptedChunk(w http.ResponseWriter, resp *http.Response, chunk *filer_pb.FileChunk) {
func (fs *FilerServer) writeEncryptedChunk(w http.ResponseWriter, resp *http.Response, entry *filer2.Entry) {
chunk := entry.Chunks[0]
encryptedData, err := ioutil.ReadAll(resp.Body)
if err != nil {
glog.V(1).Infof("read encrypted %s failed, err: %v", chunk.FileId, err)
@@ -157,6 +158,8 @@ func (fs *FilerServer) writeEncryptedChunk(w http.ResponseWriter, resp *http.Res
w.WriteHeader(http.StatusNotFound)
return
}
w.Header().Set("Content-Length", fmt.Sprintf("%d", chunk.Size))
w.WriteHeader(resp.StatusCode)
w.Write(decryptedData)
}

19
weed/server/filer_server_handlers_write.go
View File

@@ -90,10 +90,22 @@ func (fs *FilerServer) PostHandler(w http.ResponseWriter, r *http.Request) {
return
}
if fs.option.Cipher {
reply, err := fs.encrypt(ctx, w, r, replication, collection, dataCenter)
if err != nil {
writeJsonError(w, r, http.StatusInternalServerError, err)
} else if reply != nil {
writeJsonQuiet(w, r, http.StatusCreated, reply)
}
return
}
fileId, urlLocation, auth, err := fs.assignNewFileInfo(w, r, replication, collection, dataCenter)
if err != nil || fileId == "" || urlLocation == "" {
glog.V(0).Infof("fail to allocate volume for %s, collection:%s, datacenter:%s", r.URL.Path, collection, dataCenter)
writeJsonError(w, r, http.StatusInternalServerError, fmt.Errorf("fail to allocate volume for %s, collection:%s, datacenter:%s", r.URL.Path, collection, dataCenter))
return
}
@@ -134,7 +146,7 @@ func (fs *FilerServer) PostHandler(w http.ResponseWriter, r *http.Request) {
// update metadata in filer store
func (fs *FilerServer) updateFilerStore(ctx context.Context, r *http.Request, w http.ResponseWriter,
replication string, collection string, ret operation.UploadResult, fileId string) (err error) {
replication string, collection string, ret *operation.UploadResult, fileId string) (err error) {
stats.FilerRequestCounter.WithLabelValues("postStoreWrite").Inc()
start := time.Now()
@@ -198,12 +210,14 @@ func (fs *FilerServer) updateFilerStore(ctx context.Context, r *http.Request, w
}
// send request to volume server
func (fs *FilerServer) uploadToVolumeServer(r *http.Request, u *url.URL, auth security.EncodedJwt, w http.ResponseWriter, fileId string) (ret operation.UploadResult, err error) {
func (fs *FilerServer) uploadToVolumeServer(r *http.Request, u *url.URL, auth security.EncodedJwt, w http.ResponseWriter, fileId string) (ret *operation.UploadResult, err error) {
stats.FilerRequestCounter.WithLabelValues("postUpload").Inc()
start := time.Now()
defer func() { stats.FilerRequestHistogram.WithLabelValues("postUpload").Observe(time.Since(start).Seconds()) }()
ret = &operation.UploadResult{}
request := &http.Request{
Method: r.Method,
URL: u,
@@ -215,6 +229,7 @@ func (fs *FilerServer) uploadToVolumeServer(r *http.Request, u *url.URL, auth se
Host: r.Host,
ContentLength: r.ContentLength,
}
if auth != "" {
request.Header.Set("Authorization", "BEARER "+string(auth))
}

28
weed/server/filer_server_handlers_write_autochunk.go
View File

@@ -103,33 +103,35 @@ func (fs *FilerServer) doAutoChunk(ctx context.Context, w http.ResponseWriter, r
// upload the chunk to the volume server
chunkName := fileName + "_chunk_" + strconv.FormatInt(int64(len(fileChunks)+1), 10)
uploadedSize, uploadErr := fs.doUpload(urlLocation, w, r, limitedReader, chunkName, "", fileId, auth)
uploadResult, uploadErr := fs.doUpload(urlLocation, w, r, limitedReader, chunkName, "", fileId, auth)
if uploadErr != nil {
return nil, uploadErr
}
// if last chunk exhausted the reader exactly at the border
if uploadedSize == 0 {
if uploadResult.Size == 0 {
break
}
// Save to chunk manifest structure
fileChunks = append(fileChunks,
&filer_pb.FileChunk{
FileId: fileId,
Offset: chunkOffset,
Size: uint64(uploadedSize),
Mtime: time.Now().UnixNano(),
FileId: fileId,
Offset: chunkOffset,
Size: uint64(uploadResult.Size),
Mtime: time.Now().UnixNano(),
ETag: uploadResult.ETag,
CipherKey: uploadResult.CipherKey,
},
)
glog.V(4).Infof("uploaded %s chunk %d to %s [%d,%d) of %d", fileName, len(fileChunks), fileId, chunkOffset, chunkOffset+int64(uploadedSize), contentLength)
glog.V(4).Infof("uploaded %s chunk %d to %s [%d,%d) of %d", fileName, len(fileChunks), fileId, chunkOffset, chunkOffset+int64(uploadResult.Size), contentLength)
// reset variables for the next chunk
chunkOffset = chunkOffset + int64(uploadedSize)
chunkOffset = chunkOffset + int64(uploadResult.Size)
// if last chunk was not at full chunk size, but already exhausted the reader
if uploadedSize < int64(chunkSize) {
if int64(uploadResult.Size) < int64(chunkSize) {
break
}
}
@@ -174,7 +176,7 @@ func (fs *FilerServer) doAutoChunk(ctx context.Context, w http.ResponseWriter, r
}
func (fs *FilerServer) doUpload(urlLocation string, w http.ResponseWriter, r *http.Request,
limitedReader io.Reader, fileName string, contentType string, fileId string, auth security.EncodedJwt) (size int64, err error) {
limitedReader io.Reader, fileName string, contentType string, fileId string, auth security.EncodedJwt) (*operation.UploadResult, error) {
stats.FilerRequestCounter.WithLabelValues("postAutoChunkUpload").Inc()
start := time.Now()
@@ -182,9 +184,5 @@ func (fs *FilerServer) doUpload(urlLocation string, w http.ResponseWriter, r *ht
stats.FilerRequestHistogram.WithLabelValues("postAutoChunkUpload").Observe(time.Since(start).Seconds())
}()
uploadResult, uploadError := operation.Upload(urlLocation, fileName, fs.option.Cipher, limitedReader, false, contentType, nil, auth)
if uploadError != nil {
return 0, uploadError
}
return int64(uploadResult.Size), nil
return operation.Upload(urlLocation, fileName, fs.option.Cipher, limitedReader, false, contentType, nil, auth)
}

103
weed/server/filer_server_handlers_write_cipher.go Normal file
View File

@@ -0,0 +1,103 @@
package weed_server
import (
"bytes"
"context"
"fmt"
"net/http"
"strings"
"time"
"github.com/chrislusf/seaweedfs/weed/filer2"
"github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/operation"
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
"github.com/chrislusf/seaweedfs/weed/storage/needle"
"github.com/chrislusf/seaweedfs/weed/util"
)
// handling single chunk POST or PUT upload
func (fs *FilerServer) encrypt(ctx context.Context, w http.ResponseWriter, r *http.Request,
replication string, collection string, dataCenter string) (filerResult *FilerPostResult, err error) {
fileId, urlLocation, auth, err := fs.assignNewFileInfo(w, r, replication, collection, dataCenter)
if err != nil || fileId == "" || urlLocation == "" {
return nil, fmt.Errorf("fail to allocate volume for %s, collection:%s, datacenter:%s", r.URL.Path, collection, dataCenter)
}
glog.V(4).Infof("write %s to %v", r.URL.Path, urlLocation)
// Note: gzip(cipher(data)), cipher data first, then gzip
sizeLimit := int64(fs.option.MaxMB) * 1024 * 1024
pu, err := needle.ParseUpload(r, sizeLimit)
data := pu.Data
uncompressedData := pu.Data
cipherKey := util.GenCipherKey()
if pu.IsGzipped {
uncompressedData = pu.UncompressedData
data, err = util.Encrypt(pu.UncompressedData, cipherKey)
if err != nil {
return nil, fmt.Errorf("encrypt input: %v", err)
}
}
if pu.MimeType == "" {
pu.MimeType = http.DetectContentType(uncompressedData)
}
uploadResult, uploadError := operation.Upload(urlLocation, pu.FileName, true, bytes.NewReader(data), pu.IsGzipped, "", pu.PairMap, auth)
if uploadError != nil {
return nil, fmt.Errorf("upload to volume server: %v", uploadError)
}
// Save to chunk manifest structure
fileChunks := []*filer_pb.FileChunk{
{
FileId: fileId,
Offset: 0,
Size: uint64(uploadResult.Size),
Mtime: time.Now().UnixNano(),
ETag: uploadResult.ETag,
CipherKey: uploadResult.CipherKey,
},
}
path := r.URL.Path
if strings.HasSuffix(path, "/") {
if pu.FileName != "" {
path += pu.FileName
}
}
entry := &filer2.Entry{
FullPath: filer2.FullPath(path),
Attr: filer2.Attr{
Mtime: time.Now(),
Crtime: time.Now(),
Mode: 0660,
Uid: OS_UID,
Gid: OS_GID,
Replication: replication,
Collection: collection,
TtlSec: int32(util.ParseInt(r.URL.Query().Get("ttl"), 0)),
Mime: pu.MimeType,
},
Chunks: fileChunks,
}
filerResult = &FilerPostResult{
Name: pu.FileName,
Size: int64(pu.OriginalDataSize),
}
if dbErr := fs.filer.CreateEntry(ctx, entry, false); dbErr != nil {
fs.filer.DeleteChunks(entry.Chunks)
err = dbErr
filerResult.Error = dbErr.Error()
return
}
return
}