worm grace period and retention time support (#6404)

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

2 files changed, 45 insertions, 11 deletions

diff --git a/weed/server/filer_server_handlers_write.go b/weed/server/filer_server_handlers_write.go
index 75fd5984e..82880c2ac 100644
--- a/weed/server/filer_server_handlers_write.go
+++ b/weed/server/filer_server_handlers_write.go
@@ -160,8 +160,11 @@ func (fs *FilerServer) move(ctx context.Context, w http.ResponseWriter, r *http.
 		return
 	}
 
-	rule := fs.filer.FilerConf.MatchStorageRule(src)
-	if rule.Worm {
+	wormEnforced, err := fs.wormEnforcedForEntry(ctx, src)
+	if err != nil {
+		writeJsonError(w, r, http.StatusInternalServerError, err)
+		return
+	} else if wormEnforced {
 		// you cannot move a worm file or directory
 		err = fmt.Errorf("cannot move write-once entry from '%s' to '%s': operation not permitted", src, dst)
 		writeJsonError(w, r, http.StatusForbidden, err)
@@ -218,13 +221,16 @@ 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 {
+	wormEnforced, err := fs.wormEnforcedForEntry(context.TODO(), objectPath)
+	if err != nil {
+		writeJsonError(w, r, http.StatusInternalServerError, err)
+		return
+	} else if wormEnforced {
 		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)
+	err = fs.filer.DeleteEntryMetaAndData(context.Background(), util.FullPath(objectPath), isRecursive, ignoreRecursiveError, !skipChunkDeletion, false, nil, 0)
 	if err != nil && err != filer_pb.ErrNotFound {
 		glog.V(1).Infoln("deleting", objectPath, ":", err.Error())
 		writeJsonError(w, r, http.StatusInternalServerError, err)
diff --git a/weed/server/filer_server_handlers_write_autochunk.go b/weed/server/filer_server_handlers_write_autochunk.go
index d68849fe7..b0af7be4b 100644
--- a/weed/server/filer_server_handlers_write_autochunk.go
+++ b/weed/server/filer_server_handlers_write_autochunk.go
@@ -164,22 +164,50 @@ func isS3Request(r *http.Request) bool {
 
 func (fs *FilerServer) checkPermissions(ctx context.Context, r *http.Request, fileName string) error {
 	fullPath := fs.fixFilePath(ctx, r, fileName)
+	enforced, err := fs.wormEnforcedForEntry(ctx, fullPath)
+	if err != nil {
+		return err
+	} else if enforced {
+		// you cannot change a worm file
+		return errors.New("operation not permitted")
+	}
+
+	return nil
+}
+
+func (fs *FilerServer) wormEnforcedForEntry(ctx context.Context, fullPath string) (bool, error) {
 	rule := fs.filer.FilerConf.MatchStorageRule(fullPath)
 	if !rule.Worm {
-		return nil
+		return false, nil
 	}
 
-	_, err := fs.filer.FindEntry(ctx, util.FullPath(fullPath))
+	entry, err := fs.filer.FindEntry(ctx, util.FullPath(fullPath))
 	if err != nil {
 		if errors.Is(err, filer_pb.ErrNotFound) {
-			return nil
+			return false, nil
 		}
 
-		return err
+		return false, err
+	}
+
+	// worm is not enforced
+	if entry.WORMEnforcedAtTsNs == 0 {
+		return false, nil
+	}
+
+	// worm will never expire
+	if rule.WormRetentionTimeSeconds == 0 {
+		return true, nil
+	}
+
+	enforcedAt := time.Unix(0, entry.WORMEnforcedAtTsNs)
+
+	// worm is expired
+	if time.Now().Sub(enforcedAt).Seconds() >= float64(rule.WormRetentionTimeSeconds) {
+		return false, nil
 	}
 
-	// you cannot change an existing file in Worm mode
-	return errors.New("operation not permitted")
+	return true, nil
 }
 
 func (fs *FilerServer) fixFilePath(ctx context.Context, r *http.Request, fileName string) string {