feat(decomposedfs): check checksum in WriteChunk

Signed-off-by: jkoberg <jkoberg@owncloud.com>

changelog/unreleased/tusd-checksums.md

@@ -0,0 +1,5 @@
+Enhancement: Tusd PATCH checksums
+
+Check checksums also on chunked uploads during PATCH requests
+
+https://github.com/cs3org/reva/pull/4807

pkg/rhttp/datatx/manager/tus/tus.go

@@ -174,6 +174,11 @@ func (m *manager) Handler(fs storage.FS) (http.Handler, error) {
  }()
  // set etag, mtime and file id
  setHeaders(fs, w, r)
+ // set checksum
+ if v := r.Header.Get("Upload-Checksum"); v != "" {
+ ctx := context.WithValue(r.Context(), "checksum", v)
+ r = r.WithContext(ctx)
+ }
  handler.PatchFile(w, r)
  case "DELETE":
  handler.DelFile(w, r)

pkg/storage/utils/decomposedfs/node/node.go

@@ -1354,10 +1354,6 @@ func enoughDiskSpace(path string, fileSize uint64) bool {
 
 // CalculateChecksums calculates the sha1, md5 and adler32 checksums of a file
 func CalculateChecksums(ctx context.Context, path string) (hash.Hash, hash.Hash, hash.Hash32, error) {
- sha1h := sha1.New()
- md5h := md5.New()
- adler32h := adler32.New()
-
  _, subspan := tracer.Start(ctx, "os.Open")
  f, err := os.Open(path)
  subspan.End()
@@ -1366,11 +1362,20 @@ func CalculateChecksums(ctx context.Context, path string) (hash.Hash, hash.Hash,
  }
  defer f.Close()
 
- r1 := io.TeeReader(f, sha1h)
+ return CalculateChecksumsFromReader(ctx, f)
+}
+
+// CalculateChecksumsFromReader calculates the sha1, md5 and adler32 checksums of a io.Reader
+func CalculateChecksumsFromReader(ctx context.Context, r io.Reader) (hash.Hash, hash.Hash, hash.Hash32, error) {
+ sha1h := sha1.New()
+ md5h := md5.New()
+ adler32h := adler32.New()
+
+ r1 := io.TeeReader(r, sha1h)
  r2 := io.TeeReader(r1, md5h)
 
- _, subspan = tracer.Start(ctx, "io.Copy")
- _, err = io.Copy(adler32h, r2)
+ _, subspan := tracer.Start(ctx, "io.Copy")
+ _, err := io.Copy(adler32h, r2)
  subspan.End()
  if err != nil {
  return nil, nil, nil, err

pkg/storage/utils/decomposedfs/upload/upload.go

@@ -19,6 +19,7 @@
 package upload
 
 import (
+ "bytes"
  "context"
  "encoding/hex"
  "fmt"
@@ -57,6 +58,22 @@ var defaultFilePerm = os.FileMode(0664)
 func (session *OcisSession) WriteChunk(ctx context.Context, offset int64, src io.Reader) (int64, error) {
  ctx, span := tracer.Start(session.Context(ctx), "WriteChunk")
  defer span.End()
+
+ // calculate checksum here
+ if checksum, ok := ctx.Value("checksum").(string); ok {
+ // we need to copy the contents into memory so we can write it to disk later
+ b := bytes.NewBuffer(nil)
+ sha1, md5, adler32, err := node.CalculateChecksumsFromReader(ctx, io.TeeReader(src, b))
+ if err != nil {
+ return 0, err
+ }
+
+ if err := verifyChecksum(checksum, sha1, md5, adler32); err != nil {
+ return 0, err
+ }
+ src = b
+ }
+
  _, subspan := tracer.Start(ctx, "os.OpenFile")
  file, err := os.OpenFile(session.binPath(), os.O_WRONLY|os.O_APPEND, defaultFilePerm)
  subspan.End()
@@ -65,10 +82,6 @@ func (session *OcisSession) WriteChunk(ctx context.Context, offset int64, src io
  }
  defer file.Close()
 
- // calculate cheksum here? needed for the TUS checksum extension. https://tus.io/protocols/resumable-upload.html#checksum
- // TODO but how do we get the `Upload-Checksum`? WriteChunk() only has a context, offset and the reader ...
- // It is sent with the PATCH request, well or in the POST when the creation-with-upload extension is used
- // but the tus handler uses a context.Background() so we cannot really check the header and put it in the context ...
  _, subspan = tracer.Start(ctx, "io.Copy")
  n, err := io.Copy(file, src)
  subspan.End()
@@ -116,21 +129,7 @@ func (session *OcisSession) FinishUpload(ctx context.Context) error {
  // compare if they match the sent checksum
  // TODO the tus checksum extension would do this on every chunk, but I currently don't see an easy way to pass in the requested checksum. for now we do it in FinishUpload which is also called for chunked uploads
  if session.info.MetaData["checksum"] != "" {
- var err error
- parts := strings.SplitN(session.info.MetaData["checksum"], " ", 2)
- if len(parts) != 2 {
- return errtypes.BadRequest("invalid checksum format. must be '[algorithm] [checksum]'")
- }
- switch parts[0] {
- case "sha1":
- err = checkHash(parts[1], sha1h)
- case "md5":
- err = checkHash(parts[1], md5h)
- case "adler32":
- err = checkHash(parts[1], adler32h)
- default:
- err = errtypes.BadRequest("unsupported checksum algorithm: " + parts[0])
- }
+ err := verifyChecksum(session.info.MetaData["checksum"], sha1h, md5h, adler32h)
  if err != nil {
  session.store.Cleanup(ctx, session, true, false, false)
  return err
@@ -376,3 +375,20 @@ func joinurl(paths ...string) string {
 
  return s.String()
 }
+
+func verifyChecksum(checksum string, sha1h hash.Hash, md5h hash.Hash, adler32h hash.Hash32) error {
+ parts := strings.SplitN(checksum, " ", 2)
+ if len(parts) != 2 {
+ return errtypes.BadRequest("invalid checksum format. must be '[algorithm] [checksum]'")
+ }
+ switch strings.ToLower(parts[0]) {
+ case "sha1":
+ return checkHash(parts[1], sha1h)
+ case "md5":
+ return checkHash(parts[1], md5h)
+ case "adler32":
+ return checkHash(parts[1], adler32h)
+ default:
+ return errtypes.BadRequest("unsupported checksum algorithm: " + parts[0])
+ }
+}