improve worm support (#5983)

* improve worm support

Signed-off-by: lou <alex1988@outlook.com>

* worm mode in filer

Signed-off-by: lou <alex1988@outlook.com>

* update after review

Signed-off-by: lou <alex1988@outlook.com>

* update after review

Signed-off-by: lou <alex1988@outlook.com>

* move to fs configure

Signed-off-by: lou <alex1988@outlook.com>

* remove flag

Signed-off-by: lou <alex1988@outlook.com>

* update after review

Signed-off-by: lou <alex1988@outlook.com>

* support worm hardlink

Signed-off-by: lou <alex1988@outlook.com>

* update after review

Signed-off-by: lou <alex1988@outlook.com>

* typo

Signed-off-by: lou <alex1988@outlook.com>

* sync filer conf

Signed-off-by: lou <alex1988@outlook.com>

---------

Signed-off-by: lou <alex1988@outlook.com>

2 files changed, 59 insertions, 19 deletions

diff --git a/weed/server/filer_server_handlers_write.go b/weed/server/filer_server_handlers_write.go
index f0f756e34..4fc10c26c 100644
--- a/weed/server/filer_server_handlers_write.go
+++ b/weed/server/filer_server_handlers_write.go
@@ -70,7 +70,6 @@ func (fs *FilerServer) assignNewFileInfo(so *operation.StorageOption) (fileId, u
 }
 
 func (fs *FilerServer) PostHandler(w http.ResponseWriter, r *http.Request, contentLength int64) {
-
 	ctx := context.Background()
 
 	destination := r.RequestURI
@@ -197,7 +196,6 @@ func (fs *FilerServer) move(ctx context.Context, w http.ResponseWriter, r *http.
 // curl -X DELETE http://localhost:8888/path/to?recursive=true&ignoreRecursiveError=true
 // curl -X DELETE http://localhost:8888/path/to?recursive=true&skipChunkDeletion=true
 func (fs *FilerServer) DeleteHandler(w http.ResponseWriter, r *http.Request) {
-
 	isRecursive := r.FormValue("recursive") == "true"
 	if !isRecursive && fs.option.recursiveDelete {
 		if r.FormValue("recursive") != "false" {
@@ -212,6 +210,12 @@ func (fs *FilerServer) DeleteHandler(w http.ResponseWriter, r *http.Request) {
 		objectPath = objectPath[0 : len(objectPath)-1]
 	}
 
+	rule := fs.filer.FilerConf.MatchStorageRule(objectPath)
+	if rule.Worm {
+		writeJsonError(w, r, http.StatusForbidden, errors.New("operation not permitted"))
+		return
+	}
+
 	err := fs.filer.DeleteEntryMetaAndData(context.Background(), util.FullPath(objectPath), isRecursive, ignoreRecursiveError, !skipChunkDeletion, false, nil, 0)
 	if err != nil {
 		if err == filer_pb.ErrNotFound {
diff --git a/weed/server/filer_server_handlers_write_autochunk.go b/weed/server/filer_server_handlers_write_autochunk.go
index 1c7ed0c3c..cce64f74d 100644
--- a/weed/server/filer_server_handlers_write_autochunk.go
+++ b/weed/server/filer_server_handlers_write_autochunk.go
@@ -5,8 +5,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	//"github.com/seaweedfs/seaweedfs/weed/s3api"
-	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 	"io"
 	"net/http"
 	"os"
@@ -19,6 +17,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/operation"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
 	"github.com/seaweedfs/seaweedfs/weed/util"
 )
@@ -49,7 +48,9 @@ func (fs *FilerServer) autoChunk(ctx context.Context, w http.ResponseWriter, r *
 		reply, md5bytes, err = fs.doPutAutoChunk(ctx, w, r, chunkSize, contentLength, so)
 	}
 	if err != nil {
-		if strings.HasPrefix(err.Error(), "read input:") || err.Error() == io.ErrUnexpectedEOF.Error() {
+		if err.Error() == "operation not permitted" {
+			writeJsonError(w, r, http.StatusForbidden, err)
+		} else if strings.HasPrefix(err.Error(), "read input:") || err.Error() == io.ErrUnexpectedEOF.Error() {
 			writeJsonError(w, r, util.HttpStatusCancelled, err)
 		} else if strings.HasSuffix(err.Error(), "is a file") || strings.HasSuffix(err.Error(), "already exists") {
 			writeJsonError(w, r, http.StatusConflict, err)
@@ -85,6 +86,10 @@ func (fs *FilerServer) doPostAutoChunk(ctx context.Context, w http.ResponseWrite
 		contentType = ""
 	}
 
+	if err := fs.checkPermissions(ctx, r, fileName); err != nil {
+		return nil, nil, err
+	}
+
 	if so.SaveInside {
 		buf := bufPool.Get().(*bytes.Buffer)
 		buf.Reset()
@@ -121,6 +126,10 @@ func (fs *FilerServer) doPutAutoChunk(ctx context.Context, w http.ResponseWriter
 		contentType = ""
 	}
 
+	if err := fs.checkPermissions(ctx, r, ""); err != nil {
+		return nil, nil, err
+	}
+
 	fileChunks, md5Hash, chunkOffset, err, smallContent := fs.uploadReaderToChunks(w, r, r.Body, chunkSize, fileName, contentType, contentLength, so)
 	if err != nil {
 		return nil, nil, err
@@ -152,35 +161,62 @@ func isS3Request(r *http.Request) bool {
 	return r.Header.Get(s3_constants.AmzAuthType) != "" || r.Header.Get("X-Amz-Date") != ""
 }
 
-func (fs *FilerServer) saveMetaData(ctx context.Context, r *http.Request, fileName string, contentType string, so *operation.StorageOption, md5bytes []byte, fileChunks []*filer_pb.FileChunk, chunkOffset int64, content []byte) (filerResult *FilerPostResult, replyerr error) {
-
-	// detect file mode
-	modeStr := r.URL.Query().Get("mode")
-	if modeStr == "" {
-		modeStr = "0660"
+func (fs *FilerServer) checkPermissions(ctx context.Context, r *http.Request, fileName string) error {
+	fullPath := fs.fixFilePath(ctx, r, fileName)
+	rule := fs.filer.FilerConf.MatchStorageRule(fullPath)
+	if !rule.Worm {
+		return nil
 	}
-	mode, err := strconv.ParseUint(modeStr, 8, 32)
+
+	_, err := fs.filer.FindEntry(ctx, util.FullPath(fullPath))
 	if err != nil {
-		glog.Errorf("Invalid mode format: %s, use 0660 by default", modeStr)
-		mode = 0660
+		if errors.Is(err, filer_pb.ErrNotFound) {
+			return nil
+		}
+
+		return err
 	}
 
+	// you cannot change an existing file in Worm mode
+	return errors.New("operation not permitted")
+}
+
+func (fs *FilerServer) fixFilePath(ctx context.Context, r *http.Request, fileName string) string {
 	// fix the path
-	path := r.URL.Path
-	if strings.HasSuffix(path, "/") {
+	fullPath := r.URL.Path
+	if strings.HasSuffix(fullPath, "/") {
 		if fileName != "" {
-			path += fileName
+			fullPath += fileName
 		}
 	} else {
 		if fileName != "" {
-			if possibleDirEntry, findDirErr := fs.filer.FindEntry(ctx, util.FullPath(path)); findDirErr == nil {
+			if possibleDirEntry, findDirErr := fs.filer.FindEntry(ctx, util.FullPath(fullPath)); findDirErr == nil {
 				if possibleDirEntry.IsDirectory() {
-					path += "/" + fileName
+					fullPath += "/" + fileName
 				}
 			}
 		}
 	}
 
+	return fullPath
+}
+
+func (fs *FilerServer) saveMetaData(ctx context.Context, r *http.Request, fileName string, contentType string, so *operation.StorageOption, md5bytes []byte, fileChunks []*filer_pb.FileChunk, chunkOffset int64, content []byte) (filerResult *FilerPostResult, replyerr error) {
+
+	// detect file mode
+	modeStr := r.URL.Query().Get("mode")
+	if modeStr == "" {
+		modeStr = "0660"
+	}
+	mode, err := strconv.ParseUint(modeStr, 8, 32)
+	if err != nil {
+		glog.Errorf("Invalid mode format: %s, use 0660 by default", modeStr)
+		mode = 0660
+	}
+
+	// fix the path
+	path := fs.fixFilePath(ctx, r, fileName)
+
 	var entry *filer.Entry
 	var newChunks []*filer_pb.FileChunk
 	var mergedChunks []*filer_pb.FileChunk