implement if-match check in storageprovider

changelog/unreleased/storage-if-match.md

@@ -0,0 +1,5 @@
+Enhancement: Add an if-match check to the storage provider
+
+Implement a check for the if-match value in InitiateFileUpload to prevent overwrites of newer versions.
+
+https://github.com/cs3org/reva/pull/2547

internal/grpc/services/storageprovider/storageprovider.go

@@ -291,14 +291,38 @@ func (s *service) InitiateFileUpload(ctx context.Context, req *provider.Initiate
 		}, nil
 	}
 
+	metadata := map[string]string{}
+	ifMatch := req.GetIfMatch()
+	if ifMatch != "" {
+		sRes, err := s.Stat(ctx, &provider.StatRequest{Ref: req.Ref})
+		if err != nil {
+			return nil, err
+		}
+
+		switch sRes.Status.Code {
+		case rpc.Code_CODE_OK:
+			if sRes.Info.Etag != ifMatch {
+				return &provider.InitiateFileUploadResponse{
+					Status: status.NewFailedPrecondition(ctx, errors.New("etag doesn't match"), "etag doesn't match"),
+				}, nil
+			}
+		case rpc.Code_CODE_NOT_FOUND:
+			// Just continue with a normal upload
+		default:
+			return &provider.InitiateFileUploadResponse{
+				Status: sRes.Status,
+			}, nil
+		}
+		metadata["if-match"] = ifMatch
+	}
+
 	// FIXME these should be part of the InitiateFileUploadRequest object
 	if req.Opaque != nil {
 		if e, ok := req.Opaque.Map["lockid"]; ok && e.Decoder == "plain" {
 			ctx = ctxpkg.ContextSetLockID(ctx, string(e.Value))
 		}
 	}
 
-	metadata := map[string]string{}
 	var uploadLength int64
 	if req.Opaque != nil && req.Opaque.Map != nil {
 		if req.Opaque.Map["Upload-Length"] != nil {

pkg/rhttp/datatx/manager/simple/simple.go

@@ -91,6 +91,8 @@ func (m *manager) Handler(fs storage.FS) (http.Handler, error) {
 				w.WriteHeader(http.StatusUnauthorized)
 			case errtypes.InsufficientStorage:
 				w.WriteHeader(http.StatusInsufficientStorage)
+			case errtypes.PreconditionFailed:
+				w.WriteHeader(http.StatusPreconditionFailed)
 			default:
 				sublog.Error().Err(v).Msg("error uploading file")
 				w.WriteHeader(http.StatusInternalServerError)

pkg/rhttp/datatx/manager/spaces/spaces.go

@@ -103,6 +103,8 @@ func (m *manager) Handler(fs storage.FS) (http.Handler, error) {
 				w.WriteHeader(http.StatusUnauthorized)
 			case errtypes.InsufficientStorage:
 				w.WriteHeader(http.StatusInsufficientStorage)
+			case errtypes.PreconditionFailed:
+				w.WriteHeader(http.StatusPreconditionFailed)
 			default:
 				sublog.Error().Err(v).Msg("error uploading file")
 				w.WriteHeader(http.StatusInternalServerError)

pkg/storage/utils/decomposedfs/upload.go

@@ -124,24 +124,27 @@ func (fs *Decomposedfs) InitiateUpload(ctx context.Context, ref *provider.Refere
 	}
 
 	if metadata != nil {
-		if metadata["mtime"] != "" {
-			info.MetaData["mtime"] = metadata["mtime"]
+		if mtime, ok := metadata["mtime"]; ok {
+			info.MetaData["mtime"] = mtime
 		}
 		if _, ok := metadata["sizedeferred"]; ok {
 			info.SizeIsDeferred = true
 		}
-		if metadata["checksum"] != "" {
-			parts := strings.SplitN(metadata["checksum"], " ", 2)
+		if checksum, ok := metadata["checksum"]; ok {
+			parts := strings.SplitN(checksum, " ", 2)
 			if len(parts) != 2 {
 				return nil, errtypes.BadRequest("invalid checksum format. must be '[algorithm] [checksum]'")
 			}
 			switch parts[0] {
 			case "sha1", "md5", "adler32":
-				info.MetaData["checksum"] = metadata["checksum"]
+				info.MetaData["checksum"] = checksum
 			default:
 				return nil, errtypes.BadRequest("unsupported checksum algorithm: " + parts[0])
 			}
 		}
+		if ifMatch, ok := metadata["if-match"]; ok {
+			info.MetaData["if-match"] = ifMatch
+		}
 	}
 
 	log.Debug().Interface("info", info).Interface("node", n).Interface("metadata", metadata).Msg("Decomposedfs: resolved filename")
@@ -538,6 +541,19 @@ func (upload *fileUpload) FinishUpload(ctx context.Context) (err error) {
 	// if target exists create new version
 	versionsPath := ""
 	if fi, err = os.Stat(targetPath); err == nil {
+		// When the if-match header was set we need to check if the
+		// etag still matches before finishing the upload.
+		if ifMatch, ok := upload.info.MetaData["if-match"]; ok {
+			var targetEtag string
+			targetEtag, err = node.CalculateEtag(n.ID, fi.ModTime())
+			if err != nil {
+				return errtypes.InternalError(err.Error())
+			}
+			if ifMatch != targetEtag {
+				return errtypes.PreconditionFailed("etag doesn't match")
+			}
+		}
+
 		// FIXME move versioning to blobs ... no need to copy all the metadata! well ... it does if we want to version metadata...
 		// versions are stored alongside the actual file, so a rename can be efficient and does not cross storage / partition boundaries
 		versionsPath = upload.fs.lu.InternalPath(n.ID + ".REV." + fi.ModTime().UTC().Format(time.RFC3339Nano))