Propagate Content-Length header from incoming PUT call to EOS PuT call

Author: jessegeens

internal/http/services/datagateway/datagateway.go

@@ -333,6 +333,7 @@ func (s *svc) doPut(w http.ResponseWriter, r *http.Request) {
 
 	copyHeader(w.Header(), httpRes.Header)
 	if httpRes.StatusCode != http.StatusOK {
+		log.Warn().Int("StatusCode", httpRes.StatusCode).Msg("Non-OK Status Code when sending request to internal data server")
 		// swallow the body and set content-length to 0 to prevent reverse proxies from trying to read from it
 		w.Header().Set("Content-Length", "0")
 		w.WriteHeader(httpRes.StatusCode)

internal/http/services/owncloud/ocdav/put.go

@@ -278,6 +278,14 @@ func (s *svc) handlePut(ctx context.Context, w http.ResponseWriter, r *http.Requ
 		httpReq.Header.Set(HeaderLockHolder, lockholder)
 	}
 
+	// We need to pass Content-Length to the storage backend (e.g. EOS).
+	// However, the Go HTTP Client library may arbitrarily modify or drop
+	// the Content-Length header, for example when it compresses the data
+	// See: https://pkg.go.dev/net/http#Request
+	// Therefore, we use another header to pass it through the internal
+	// data server
+	httpReq.Header.Set(HeaderUploadLength, strconv.FormatInt(length, 10))
+
 	// Propagate X-Disable-Versioning header
 	// Used to disable versioning for applications that do not expect this behaviour
 	// See reva#4855 for more info

pkg/eosclient/eosbinary/eosbinary.go

@@ -722,7 +722,7 @@ func (c *Client) Read(ctx context.Context, auth eosclient.Authorization, path st
 }
 
 // Write writes a stream to the mgm.
-func (c *Client) Write(ctx context.Context, auth eosclient.Authorization, path string, stream io.ReadCloser, app string, disableVersioning bool) error {
+func (c *Client) Write(ctx context.Context, auth eosclient.Authorization, path string, stream io.ReadCloser, length int64, app string, disableVersioning bool) error {
 	fd, err := os.CreateTemp(c.opt.CacheDirectory, "eoswrite-")
 	if err != nil {
 		return err

pkg/eosclient/eosclient.go

@@ -51,7 +51,7 @@ type EOSClient interface {
 	Rename(ctx context.Context, auth Authorization, oldPath, newPath string) error
 	List(ctx context.Context, auth Authorization, path string) ([]*FileInfo, error)
 	Read(ctx context.Context, auth Authorization, path string) (io.ReadCloser, error)
-	Write(ctx context.Context, auth Authorization, path string, stream io.ReadCloser, app string, disableVersioning bool) error
+	Write(ctx context.Context, auth Authorization, path string, stream io.ReadCloser, length int64, app string, disableVersioning bool) error
 	ListDeletedEntries(ctx context.Context, auth Authorization, maxentries int, from, to time.Time) ([]*DeletedEntry, error)
 	RestoreDeletedEntry(ctx context.Context, auth Authorization, key string) error
 	PurgeDeletedEntries(ctx context.Context, auth Authorization) error

pkg/eosclient/eosgrpc/eosgrpc.go

@@ -1364,11 +1364,9 @@ func (c *Client) Read(ctx context.Context, auth eosclient.Authorization, path st
 
 // Write writes a file to the mgm
 // Somehow the same considerations as Read apply.
-func (c *Client) Write(ctx context.Context, auth eosclient.Authorization, path string, stream io.ReadCloser, app string, disableVersioning bool) error {
+func (c *Client) Write(ctx context.Context, auth eosclient.Authorization, path string, stream io.ReadCloser, length int64, app string, disableVersioning bool) error {
 	log := appctx.GetLogger(ctx)
 	log.Info().Str("func", "Write").Str("uid,gid", auth.Role.UID+","+auth.Role.GID).Str("path", path).Msg("")
-	var length int64
-	length = -1
 
 	u, err := getUser(ctx)
 	if err != nil {

pkg/eosclient/eosgrpc/eoshttp.go

@@ -373,10 +373,6 @@ func (c *EOSHTTPClient) PUTFile(ctx context.Context, remoteuser string, auth eos
 	base.RawQuery = queryValues.Encode()
 	finalurl := base.String()
 
-	if err != nil {
-		log.Error().Str("func", "PUTFile").Str("url", finalurl).Str("err", err.Error()).Msg("can't create request")
-		return err
-	}
 	req, err := http.NewRequestWithContext(ctx, http.MethodPut, finalurl, nil)
 	if err != nil {
 		log.Error().Str("func", "PUTFile").Str("url", finalurl).Str("err", err.Error()).Msg("can't create request")
@@ -424,15 +420,8 @@ func (c *EOSHTTPClient) PUTFile(ctx context.Context, remoteuser string, auth eos
 			}
 			if length >= 0 {
 				log.Debug().Str("func", "PUTFile").Int64("Content-Length", length).Msg("setting header")
-				req.Header.Set("Content-Length", strconv.FormatInt(length, 10))
-			}
-			if err != nil {
-				log.Error().Str("func", "PUTFile").Str("url", loc.String()).Str("err", err.Error()).Msg("can't create redirected request")
-				return err
-			}
-			if length >= 0 {
-				log.Debug().Str("func", "PUTFile").Int64("Content-Length", length).Msg("setting header")
-				req.Header.Set("Content-Length", strconv.FormatInt(length, 10))
+				req.ContentLength = length
+				req.Header.Set("Content-Length", fmt.Sprintf("%d", length))
 			}
 
 			log.Debug().Str("func", "PUTFile").Str("location", loc.String()).Msg("redirection")

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

@@ -21,6 +21,7 @@ package simple
 import (
 	"context"
 	"net/http"
+	"strconv"
 
 	provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
 	"github.com/cs3org/reva/internal/http/services/owncloud/ocdav"
@@ -88,6 +89,14 @@ func (m *manager) Handler(fs storage.FS) (http.Handler, error) {
 				metadata["disableVersioning"] = disableVersioning
 			}
 
+			contentLength := r.Header.Get(ocdav.HeaderUploadLength)
+
+			if _, err := strconv.ParseInt(contentLength, 10, 64); err == nil {
+				metadata[ocdav.HeaderContentLength] = contentLength
+			} else {
+				w.WriteHeader(http.StatusBadRequest)
+			}
+
 			err := fs.Upload(ctx, ref, r.Body, metadata)
 			switch v := err.(type) {
 			case nil:

pkg/storage/utils/eosfs/upload.go

@@ -26,6 +26,7 @@ import (
 	"strconv"
 
 	provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
+	"github.com/cs3org/reva/internal/http/services/owncloud/ocdav"
 	"github.com/cs3org/reva/pkg/errtypes"
 	"github.com/cs3org/reva/pkg/storage/utils/chunking"
 	"github.com/pkg/errors"
@@ -93,7 +94,13 @@ func (fs *eosfs) Upload(ctx context.Context, ref *provider.Reference, r io.ReadC
 		disableVersioning = false
 	}
 
-	return fs.c.Write(ctx, auth, fn, r, app, disableVersioning)
+	contentLength := metadata[ocdav.HeaderContentLength]
+	len, err := strconv.ParseInt(contentLength, 10, 64)
+	if err != nil {
+		return errors.New("No content length specified in EOS upload, got: " + contentLength)
+	}
+
+	return fs.c.Write(ctx, auth, fn, r, len, app, disableVersioning)
 }
 
 func (fs *eosfs) InitiateUpload(ctx context.Context, ref *provider.Reference, uploadLength int64, metadata map[string]string) (map[string]string, error) {

tests/helpers/helpers.go

@@ -28,13 +28,15 @@ import (
 	"os"
 	"path/filepath"
 	"runtime"
+	"strconv"
 
 	"github.com/pkg/errors"
 
 	gatewayv1beta1 "github.com/cs3org/go-cs3apis/cs3/gateway/v1beta1"
 	rpcv1beta1 "github.com/cs3org/go-cs3apis/cs3/rpc/v1beta1"
 	provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
 	"github.com/cs3org/reva/internal/http/services/datagateway"
+	"github.com/cs3org/reva/internal/http/services/owncloud/ocdav"
 	"github.com/cs3org/reva/pkg/httpclient"
 	"github.com/cs3org/reva/pkg/storage"
 	"github.com/studio-b12/gowebdav"
@@ -105,7 +107,9 @@ func Upload(ctx context.Context, fs storage.FS, ref *provider.Reference, content
 		return errors.New("simple upload method not available")
 	}
 	uploadRef := &provider.Reference{Path: "/" + uploadID}
-	err = fs.Upload(ctx, uploadRef, io.NopCloser(bytes.NewReader(content)), map[string]string{})
+	err = fs.Upload(ctx, uploadRef, io.NopCloser(bytes.NewReader(content)), map[string]string{
+		ocdav.HeaderUploadLength: strconv.FormatInt(int64(len(content)), 10),
+	})
 	return err
 }