Repository: rclone/rclone
Pull request #3951 (2020-02-10 09:32:17)

fs: add header-download and header-upload flags
#### What is the purpose of this change?

The goal is to add two repeatable flags as follows

--header-upload "Header: String"
--header-download "Header: String"

Headers added with the --header-upload flag will get added to all Put/Update calls.
Headers added with the --header-download flag will get added to all Open calls.

In this PR I would like to lay the groundwork for these flags and complete an implementation for at least one of the backends. My immediate need is for s3/minio so I am focusing on that.

I have been following the implementation steps laid out [here](https://github.com/rclone/rclone/issues/59#issuecomment-504379064).

#### Was the change discussed in an issue or in the forum before?

https://github.com/rclone/rclone/issues/59

#### Checklist

- [X] I have read the [contribution guidelines](https://github.com/rclone/rclone/blob/master/CONTRIBUTING.md#submitting-a-pull-request).
- [x] I have added tests for all changes in this PR if appropriate.
- [x] I have added documentation for the changes if appropriate.
- [x] All commit messages are in [house style](https://github.com/rclone/rclone/blob/master/CONTRIBUTING.md#commit-messages).
- [x] I'm done, this Pull Request is ready for review :-)


tgallant: This is currently a work in progress.

So far I have:
- added the two flags
- added UploadHeaders to Update/Put options in operations.go
- added UploadHeaders on Update for the s3 backend

With the current state of this branch I can add arbitrary headers to s3 uploads. Here's some truncated output 

```shell
~/go/src/rclone $ ./rclone sync -vv ~/src_folder/ minio-test:test/dst_folder --upload-header "x-amz-meta-foo: hello world" --dump headers
2020/02/10 01:27:13 DEBUG : rclone: Version "v1.51.0-DEV" starting with parameters ["./rclone" "sync" "-vv" "/Users/tim/src_folder/" "minio-test:test/dst_folder" "--upload-header" "x-amz-meta-foo: hello world" "--dump" "headers"]
2020/02/10 01:27:13 DEBUG : Using config file from "/Users/tim/.config/rclone/rclone.conf"
2020/02/10 01:27:13 DEBUG : You have specified to dump information. Please be noted that the Accept-Encoding as shown may not be correct in the request and the response may not show Content-Encoding if the go standard libraries auto gzip encoding was in effect. In this case the body of the request will be gunzipped before showing it.

...

2020/02/10 01:27:13 DEBUG : HTTP REQUEST (req 0xc000171500)
2020/02/10 01:27:13 DEBUG : PUT /test/dst_folder/hi.txt?X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=minioadmin%2F20200210%2Fus-east-1%2Fs3%2Faws4_request&X-Amz-Date=20200210T092713Z&X-Amz-Expires=900&X-Amz-SignedHeaders=content-md5%3Bcontent-type%3Bhost%3Bx-amz-acl%3Bx-amz-meta-mtime&X-Amz-Signature=f921ab4cf0218f47d6f94420cc51c125bae350f83654c3a45c4cd8e3aee7dee8 HTTP/1.1
Host: localhost:9000
User-Agent: rclone/v1.51.0-DEV
Content-Length: 12
X-Amz-Meta-Foo: hello world
content-md5: u6fpHmYS00i48ARWt6412w==
content-type: text/plain; charset=utf-8
x-amz-acl: private
x-amz-meta-mtime: 1581284369.033759166
Accept-Encoding: gzip

...

2020/02/10 01:27:13 DEBUG : foo/bar/file.txt: MD5 = 764efa883dda1e11db47671c4a3bbd9e OK
2020/02/10 01:27:13 INFO  : foo/bar/file.txt: Copied (new)
2020/02/10 01:27:13 INFO  : Waiting for deletions to finish
2020/02/10 01:27:13 INFO  : 
Transferred:   	        15 / 15 Bytes, 100%, 119 Bytes/s, ETA 0s
Transferred:            2 / 2, 100%
Elapsed time:         0.1s

2020/02/10 01:27:13 DEBUG : 8 go routines active
2020/02/10 01:27:13 DEBUG : rclone: Version "v1.51.0-DEV" finishing with parameters ["./rclone" "sync" "-vv" "/Users/tim/src_folder/" "minio-test:test/dst_folder" "--upload-header" "x-amz-meta-foo: hello world" "--dump" "headers"]

```

How's this looking to you so far @ncw? If this is going in the right direction then I can continue to add support for these headers in operations.go and s3.go.
ncw: > How's this looking to you so far @ncw? If this is going in the right direction then I can continue to add support for these headers in operations.go and s3.go.

Great - keep going :-)
tgallant: I decided to rename these headers from `--download-header` and `--upload-header` to `--header-download` and `--header-upload`. This allows the flags to show up next to each other in the help output and I think that is useful. Let me know if there are any objections.

I've got this working for `Open` requests now too.

```shell
~/go/src/rclone $ ./rclone -vv sync --exclude ".**" minio-test:test/dst ~/to_export --header-download "X-Amz-Meta-Test1: hi" --header-download "X-Amz-Meta-Test2: there" --dump headers
2020/02/15 18:54:21 DEBUG : rclone: Version "v1.51.0-DEV" starting with parameters ["./rclone" "-vv" "sync" "--exclude" ".**" "minio-test:test/dst" "/Users/tim/to_export" "--header-download" "X-Amz-Meta-Test1: hi" "--header-download" "X-Amz-Meta-Test2: there" "--dump" "headers"]
2020/02/15 18:54:21 DEBUG : Using config file from "/Users/tim/.config/rclone/rclone.conf"

...

2020/02/15 18:54:21 DEBUG : HTTP REQUEST (req 0xc00024c200)
2020/02/15 18:54:21 DEBUG : GET /test/dst/publish.el HTTP/1.1
Host: localhost:9000
User-Agent: rclone/v1.51.0-DEV
Authorization: XXXX
X-Amz-Content-Sha256: e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855
X-Amz-Date: 20200216T025421Z
X-Amz-Meta-Test1: hi
X-Amz-Meta-Test2: there
Accept-Encoding: gzip

...

2020/02/15 18:54:21 DEBUG : publish.el: MD5 = d5bca169076d2908248265684ce7f3c7 OK
2020/02/15 18:54:21 INFO  : publish.el: Copied (new)
2020/02/15 18:54:21 INFO  : Waiting for deletions to finish
2020/02/15 18:54:21 INFO  : 
Transferred:   	       706 / 706 Bytes, 100%, 218.813 kBytes/s, ETA 0s
Transferred:            2 / 2, 100%
Elapsed time:         0.0s

2020/02/15 18:54:21 DEBUG : 8 go routines active
2020/02/15 18:54:21 DEBUG : rclone: Version "v1.51.0-DEV" finishing with parameters ["./rclone" "-vv" "sync" "--exclude" ".**" "minio-test:test/dst" "/Users/tim/to_export" "--header-download" "X-Amz-Meta-Test1: hi" "--header-download" "X-Amz-Meta-Test2: there" "--dump" "headers"]
```

I think this PR is just about done. I want to do some more testing with s3/minio as well as the `lib/rest` backends. I'll provide another update when this is ready to review.

Any thoughts on how/where to add tests for these changes @ncw?
tgallant: I did some testing with the `lib/rest` backends. It looks like all of them will work out of the box for Open requests since they are passing in the options via `rest.Opts`. However it doesn't look like any of the backends are doing this for Put or Update requests.

We'll just have to go through and pass through those options for the Put/Update requests as well. I've done this for the b2 backend [here](https://github.com/rclone/rclone/pull/3951/commits/0c7f05210e7c49d556b1ac3f6378d3bc4ad5cf4d).

```go
opts := rest.Opts{
  Method:  "POST",
  RootURL: upload.UploadURL,
  Body:    in,
  Options: options,
  ExtraHeaders: map[string]string{
    "Authorization":  upload.AuthorizationToken,
    "X-Bz-File-Name": urlEncode(o.fs.opt.Enc.FromStandardPath(bucketPath)),
    "Content-Type":   fs.MimeType(ctx, src),
    sha1Header:       calculatedSha1,
    timeHeader:       timeString(modTime),
  },
  ContentLength: &size,
}
```
Dibbyo456: @tgallant Thank you doing this work.

I was wondering if you could add flag for additional http headers for **every** http requests (such as GET, POST, PUT etc)
tgallant: Hey @ncw, these header options are now wired up and working for all of the lib/rest backends. I could use some help identifying a course of action for the remaining backends. Seems like support for this would depend on the client libraries being used per backend.

Happy to keep working on this. Let me know what you'd like to see in order for this to get merged.
ncw: Any thoughts on the above @tgallant ?
tgallant: > Any thoughts on the above?

Sorry for the delay @ncw. That all sounds good to me. I did a fresh rebase and pushed up the doc changes and the merge fix. Let me know if there is anything else.

> The http backend should work already shouldn't it?

Yes, this is working for the http backend as well.
ncw: Thank you that is looking really good!

I fixed the merge conflict (which I introduced by making `newReOpen` public!).

I'll merge this now - thank you very much for your hard work :-)

Patch:
diff --git a/backend/b2/b2.go b/backend/b2/b2.go
index c622c7ba7..d69c47f44 100644
--- a/backend/b2/b2.go
+++ b/backend/b2/b2.go
@@ -1816,6 +1816,7 @@ func (o *Object) Update(ctx context.Context, in io.Reader, src fs.ObjectInfo, op
 		Method:  "POST",
 		RootURL: upload.UploadURL,
 		Body:    in,
+		Options: options,
 		ExtraHeaders: map[string]string{
 			"Authorization":  upload.AuthorizationToken,
 			"X-Bz-File-Name": urlEncode(o.fs.opt.Enc.FromStandardPath(bucketPath)),
diff --git a/backend/box/box.go b/backend/box/box.go
index e6691ac4f..e444b9015 100644
--- a/backend/box/box.go
+++ b/backend/box/box.go
@@ -1191,7 +1191,7 @@ func (o *Object) Open(ctx context.Context, options ...fs.OpenOption) (in io.Read
 // upload does a single non-multipart upload
 //
 // This is recommended for less than 50 MB of content
-func (o *Object) upload(ctx context.Context, in io.Reader, leaf, directoryID string, modTime time.Time) (err error) {
+func (o *Object) upload(ctx context.Context, in io.Reader, leaf, directoryID string, modTime time.Time, options ...fs.OpenOption) (err error) {
 	upload := api.UploadFile{
 		Name:              o.fs.opt.Enc.FromStandardName(leaf),
 		ContentModifiedAt: api.Time(modTime),
@@ -1210,6 +1210,7 @@ func (o *Object) upload(ctx context.Context, in io.Reader, leaf, directoryID str
 		MultipartContentName:  "contents",
 		MultipartFileName:     upload.Name,
 		RootURL:               uploadURL,
+		Options:               options,
 	}
 	// If object has an ID then it is existing so create a new version
 	if o.id != "" {
@@ -1251,9 +1252,9 @@ func (o *Object) Update(ctx context.Context, in io.Reader, src fs.ObjectInfo, op
 
 	// Upload with simple or multipart
 	if size <= int64(o.fs.opt.UploadCutoff) {
-		err = o.upload(ctx, in, leaf, directoryID, modTime)
+		err = o.upload(ctx, in, leaf, directoryID, modTime, options...)
 	} else {
-		err = o.uploadMultipart(ctx, in, leaf, directoryID, size, modTime)
+		err = o.uploadMultipart(ctx, in, leaf, directoryID, size, modTime, options...)
 	}
 	return err
 }
diff --git a/backend/box/upload.go b/backend/box/upload.go
index b832bf1c3..e259740c2 100644
--- a/backend/box/upload.go
+++ b/backend/box/upload.go
@@ -54,7 +54,7 @@ func sha1Digest(digest []byte) string {
 }
 
 // uploadPart uploads a part in an upload session
-func (o *Object) uploadPart(ctx context.Context, SessionID string, offset, totalSize int64, chunk []byte, wrap accounting.WrapFn) (response *api.UploadPartResponse, err error) {
+func (o *Object) uploadPart(ctx context.Context, SessionID string, offset, totalSize int64, chunk []byte, wrap accounting.WrapFn, options ...fs.OpenOption) (response *api.UploadPartResponse, err error) {
 	chunkSize := int64(len(chunk))
 	sha1sum := sha1.Sum(chunk)
 	opts := rest.Opts{
@@ -64,6 +64,7 @@ func (o *Object) uploadPart(ctx context.Context, SessionID string, offset, total
 		ContentType:   "application/octet-stream",
 		ContentLength: &chunkSize,
 		ContentRange:  fmt.Sprintf("bytes %d-%d/%d", offset, offset+chunkSize-1, totalSize),
+		Options:       options,
 		ExtraHeaders: map[string]string{
 			"Digest": sha1Digest(sha1sum[:]),
 		},
@@ -171,7 +172,7 @@ func (o *Object) abortUpload(ctx context.Context, SessionID string) (err error)
 }
 
 // uploadMultipart uploads a file using multipart upload
-func (o *Object) uploadMultipart(ctx context.Context, in io.Reader, leaf, directoryID string, size int64, modTime time.Time) (err error) {
+func (o *Object) uploadMultipart(ctx context.Context, in io.Reader, leaf, directoryID string, size int64, modTime time.Time, options ...fs.OpenOption) (err error) {
 	// Create upload session
 	session, err := o.createUploadSession(ctx, leaf, directoryID, size)
 	if err != nil {
@@ -236,7 +237,7 @@ outer:
 			defer wg.Done()
 			defer o.fs.uploadToken.Put()
 			fs.Debugf(o, "Uploading part %d/%d offset %v/%v part size %v", part+1, session.TotalParts, fs.SizeSuffix(position), fs.SizeSuffix(size), fs.SizeSuffix(chunkSize))
-			partResponse, err := o.uploadPart(ctx, session.ID, position, size, buf, wrap)
+			partResponse, err := o.uploadPart(ctx, session.ID, position, size, buf, wrap, options...)
 			if err != nil {
 				err = errors.Wrap(err, "multipart upload failed to upload part")
 				select {
diff --git a/backend/fichier/api.go b/backend/fichier/api.go
index 73fcfc597..4454e0421 100644
--- a/backend/fichier/api.go
+++ b/backend/fichier/api.go
@@ -320,7 +320,7 @@ func (f *Fs) getUploadNode(ctx context.Context) (response *GetUploadNodeResponse
 	return response, err
 }
 
-func (f *Fs) uploadFile(ctx context.Context, in io.Reader, size int64, fileName, folderID, uploadID, node string) (response *http.Response, err error) {
+func (f *Fs) uploadFile(ctx context.Context, in io.Reader, size int64, fileName, folderID, uploadID, node string, options ...fs.OpenOption) (response *http.Response, err error) {
 	// fs.Debugf(f, "Uploading File `%s`", fileName)
 
 	fileName = f.opt.Enc.FromStandardName(fileName)
@@ -338,6 +338,7 @@ func (f *Fs) uploadFile(ctx context.Context, in io.Reader, size int64, fileName,
 		NoResponse:           true,
 		Body:                 in,
 		ContentLength:        &size,
+		Options:              options,
 		MultipartContentName: "file[]",
 		MultipartFileName:    fileName,
 		MultipartParams: map[string][]string{
diff --git a/backend/fichier/fichier.go b/backend/fichier/fichier.go
index f237af48e..3f0da19a6 100644
--- a/backend/fichier/fichier.go
+++ b/backend/fichier/fichier.go
@@ -338,7 +338,7 @@ func (f *Fs) putUnchecked(ctx context.Context, in io.Reader, remote string, size
 		return nil, err
 	}
 
-	_, err = f.uploadFile(ctx, in, size, leaf, directoryID, nodeResponse.ID, nodeResponse.URL)
+	_, err = f.uploadFile(ctx, in, size, leaf, directoryID, nodeResponse.ID, nodeResponse.URL, options...)
 	if err != nil {
 		return nil, err
 	}
diff --git a/backend/googlephotos/googlephotos.go b/backend/googlephotos/googlephotos.go
index 292bb0974..e8bc2b371 100644
--- a/backend/googlephotos/googlephotos.go
+++ b/backend/googlephotos/googlephotos.go
@@ -954,8 +954,9 @@ func (o *Object) Update(ctx context.Context, in io.Reader, src fs.ObjectInfo, op
 
 	// Upload the media item in exchange for an UploadToken
 	opts := rest.Opts{
-		Method: "POST",
-		Path:   "/uploads",
+		Method:  "POST",
+		Path:    "/uploads",
+		Options: options,
 		ExtraHeaders: map[string]string{
 			"X-Goog-Upload-File-Name": fileName,
 			"X-Goog-Upload-Protocol":  "raw",
diff --git a/backend/jottacloud/jottacloud.go b/backend/jottacloud/jottacloud.go
index 5422766ca..f8ecad41e 100644
--- a/backend/jottacloud/jottacloud.go
+++ b/backend/jottacloud/jottacloud.go
@@ -1290,6 +1290,7 @@ func (o *Object) Update(ctx context.Context, in io.Reader, src fs.ObjectInfo, op
 	opts := rest.Opts{
 		Method:       "POST",
 		Path:         "files/v1/allocate",
+		Options:      options,
 		ExtraHeaders: make(map[string]string),
 	}
 	fileDate := api.Time(src.ModTime(ctx)).APIString()
diff --git a/backend/onedrive/onedrive.go b/backend/onedrive/onedrive.go
index 10e25bb78..9c9eb1da2 100755
--- a/backend/onedrive/onedrive.go
+++ b/backend/onedrive/onedrive.go
@@ -1612,7 +1612,7 @@ func (o *Object) getPosition(ctx context.Context, url string) (pos int64, err er
 }
 
 // uploadFragment uploads a part
-func (o *Object) uploadFragment(ctx context.Context, url string, start int64, totalSize int64, chunk io.ReadSeeker, chunkSize int64) (info *api.Item, err error) {
+func (o *Object) uploadFragment(ctx context.Context, url string, start int64, totalSize int64, chunk io.ReadSeeker, chunkSize int64, options ...fs.OpenOption) (info *api.Item, err error) {
 	//	var response api.UploadFragmentResponse
 	var resp *http.Response
 	var body []byte
@@ -1625,6 +1625,7 @@ func (o *Object) uploadFragment(ctx context.Context, url string, start int64, to
 			ContentLength: &toSend,
 			ContentRange:  fmt.Sprintf("bytes %d-%d/%d", start+skip, start+chunkSize-1, totalSize),
 			Body:          chunk,
+			Options:       options,
 		}
 		_, _ = chunk.Seek(skip, io.SeekStart)
 		resp, err = o.fs.srv.Call(ctx, &opts)
@@ -1682,7 +1683,7 @@ func (o *Object) cancelUploadSession(ctx context.Context, url string) (err error
 }
 
 // uploadMultipart uploads a file using multipart upload
-func (o *Object) uploadMultipart(ctx context.Context, in io.Reader, size int64, modTime time.Time) (info *api.Item, err error) {
+func (o *Object) uploadMultipart(ctx context.Context, in io.Reader, size int64, modTime time.Time, options ...fs.OpenOption) (info *api.Item, err error) {
 	if size <= 0 {
 		return nil, errors.New("unknown-sized upload not supported")
 	}
@@ -1733,7 +1734,7 @@ func (o *Object) uploadMultipart(ctx context.Context, in io.Reader, size int64,
 		}
 		seg := readers.NewRepeatableReader(io.LimitReader(in, n))
 		fs.Debugf(o, "Uploading segment %d/%d size %d", position, size, n)
-		info, err = o.uploadFragment(ctx, uploadURL, position, size, seg, n)
+		info, err = o.uploadFragment(ctx, uploadURL, position, size, seg, n, options...)
 		if err != nil {
 			return nil, err
 		}
@@ -1746,7 +1747,7 @@ func (o *Object) uploadMultipart(ctx context.Context, in io.Reader, size int64,
 
 // Update the content of a remote file within 4MB size in one single request
 // This function will set modtime after uploading, which will create a new version for the remote file
-func (o *Object) uploadSinglepart(ctx context.Context, in io.Reader, size int64, modTime time.Time) (info *api.Item, err error) {
+func (o *Object) uploadSinglepart(ctx context.Context, in io.Reader, size int64, modTime time.Time, options ...fs.OpenOption) (info *api.Item, err error) {
 	if size < 0 || size > int64(fs.SizeSuffix(4*1024*1024)) {
 		return nil, errors.New("size passed into uploadSinglepart must be >= 0 and <= 4MiB")
 	}
@@ -1763,6 +1764,7 @@ func (o *Object) uploadSinglepart(ctx context.Context, in io.Reader, size int64,
 			Path:          "/" + drive + "/items/" + trueDirID + ":/" + rest.URLPathEscape(o.fs.opt.Enc.FromStandardName(leaf)) + ":/content",
 			ContentLength: &size,
 			Body:          in,
+			Options:       options,
 		}
 	} else {
 		opts = rest.Opts{
@@ -1770,6 +1772,7 @@ func (o *Object) uploadSinglepart(ctx context.Context, in io.Reader, size int64,
 			Path:          "/root:/" + rest.URLPathEscape(o.srvPath()) + ":/content",
 			ContentLength: &size,
 			Body:          in,
+			Options:       options,
 		}
 	}
 
@@ -1811,9 +1814,9 @@ func (o *Object) Update(ctx context.Context, in io.Reader, src fs.ObjectInfo, op
 
 	var info *api.Item
 	if size > 0 {
-		info, err = o.uploadMultipart(ctx, in, size, modTime)
+		info, err = o.uploadMultipart(ctx, in, size, modTime, options...)
 	} else if size == 0 {
-		info, err = o.uploadSinglepart(ctx, in, size, modTime)
+		info, err = o.uploadSinglepart(ctx, in, size, modTime, options...)
 	} else {
 		return errors.New("unknown-sized upload not supported")
 	}
diff --git a/backend/opendrive/opendrive.go b/backend/opendrive/opendrive.go
index d5f41c817..3f8c05f83 100644
--- a/backend/opendrive/opendrive.go
+++ b/backend/opendrive/opendrive.go
@@ -687,8 +687,9 @@ func (f *Fs) Put(ctx context.Context, in io.Reader, src fs.ObjectInfo, options .
 				Name:      leaf,
 			}
 			opts := rest.Opts{
-				Method: "POST",
-				Path:   "/upload/create_file.json",
+				Method:  "POST",
+				Options: options,
+				Path:    "/upload/create_file.json",
 			}
 			resp, err = o.fs.srv.CallJSON(ctx, &opts, &createFileData, &response)
 			return o.fs.shouldRetry(resp, err)
@@ -970,8 +971,9 @@ func (o *Object) Update(ctx context.Context, in io.Reader, src fs.ObjectInfo, op
 		openUploadData := openUpload{SessionID: o.fs.session.SessionID, FileID: o.id, Size: size}
 		// fs.Debugf(nil, "PreOpen: %#v", openUploadData)
 		opts := rest.Opts{
-			Method: "POST",
-			Path:   "/upload/open_file_upload.json",
+			Method:  "POST",
+			Options: options,
+			Path:    "/upload/open_file_upload.json",
 		}
 		resp, err := o.fs.srv.CallJSON(ctx, &opts, &openUploadData, &openResponse)
 		return o.fs.shouldRetry(resp, err)
diff --git a/backend/pcloud/pcloud.go b/backend/pcloud/pcloud.go
index 51c9ec209..c2735014d 100644
--- a/backend/pcloud/pcloud.go
+++ b/backend/pcloud/pcloud.go
@@ -1080,6 +1080,7 @@ func (o *Object) Update(ctx context.Context, in io.Reader, src fs.ObjectInfo, op
 		ContentLength:    &size,
 		Parameters:       url.Values{},
 		TransferEncoding: []string{"identity"}, // pcloud doesn't like chunked encoding
+		Options:          options,
 	}
 	leaf = o.fs.opt.Enc.FromStandardName(leaf)
 	opts.Parameters.Set("filename", leaf)
diff --git a/backend/premiumizeme/premiumizeme.go b/backend/premiumizeme/premiumizeme.go
index f090dd0b5..e89fab1e7 100644
--- a/backend/premiumizeme/premiumizeme.go
+++ b/backend/premiumizeme/premiumizeme.go
@@ -517,7 +517,7 @@ func (f *Fs) Put(ctx context.Context, in io.Reader, src fs.ObjectInfo, options .
 		return existingObj, existingObj.Update(ctx, in, src, options...)
 	case fs.ErrorObjectNotFound:
 		// Not found so create it
-		return f.PutUnchecked(ctx, in, src)
+		return f.PutUnchecked(ctx, in, src, options...)
 	default:
 		return nil, err
 	}
@@ -1002,6 +1002,7 @@ func (o *Object) Update(ctx context.Context, in io.Reader, src fs.ObjectInfo, op
 		Method:     "POST",
 		Path:       "/folder/uploadinfo",
 		Parameters: o.fs.baseParams(),
+		Options:    options,
 		MultipartParams: url.Values{
 			"id": {directoryID},
 		},
diff --git a/backend/s3/s3.go b/backend/s3/s3.go
index 1353fd666..feb8500c1 100644
--- a/backend/s3/s3.go
+++ b/backend/s3/s3.go
@@ -2127,22 +2127,26 @@ func (o *Object) Open(ctx context.Context, options ...fs.OpenOption) (in io.Read
 	if o.fs.opt.SSECustomerKeyMD5 != "" {
 		req.SSECustomerKeyMD5 = &o.fs.opt.SSECustomerKeyMD5
 	}
+	httpReq, resp := o.fs.c.GetObjectRequest(&req)
 	fs.FixRangeOption(options, o.bytes)
 	for _, option := range options {
 		switch option.(type) {
 		case *fs.RangeOption, *fs.SeekOption:
 			_, value := option.Header()
 			req.Range = &value
+		case *fs.HTTPOption:
+			key, value := option.Header()
+			httpReq.HTTPRequest.Header.Add(key, value)
 		default:
 			if option.Mandatory() {
 				fs.Logf(o, "Unsupported mandatory option: %v", option)
 			}
 		}
 	}
-	var resp *s3.GetObjectOutput
 	err = o.fs.pacer.Call(func() (bool, error) {
 		var err error
-		resp, err = o.fs.c.GetObjectWithContext(ctx, &req)
+		httpReq.HTTPRequest = httpReq.HTTPRequest.WithContext(ctx)
+		err = httpReq.Send()
 		return o.fs.shouldRetry(err)
 	})
 	if err, ok := err.(awserr.RequestFailure); ok {
@@ -2449,6 +2453,18 @@ func (o *Object) Update(ctx context.Context, in io.Reader, src fs.ObjectInfo, op
 		httpReq.Header = headers
 		httpReq.ContentLength = size
 
+		for _, option := range options {
+			switch option.(type) {
+			case *fs.HTTPOption:
+				key, value := option.Header()
+				httpReq.Header.Add(key, value)
+			default:
+				if option.Mandatory() {
+					fs.Logf(o, "Unsupported mandatory option: %v", option)
+				}
+			}
+		}
+
 		err = o.fs.pacer.CallNoRetry(func() (bool, error) {
 			resp, err := o.fs.srv.Do(httpReq)
 			if err != nil {
diff --git a/backend/sharefile/sharefile.go b/backend/sharefile/sharefile.go
index 2be175003..798bb04c5 100644
--- a/backend/sharefile/sharefile.go
+++ b/backend/sharefile/sharefile.go
@@ -1429,8 +1429,9 @@ func (o *Object) Update(ctx context.Context, in io.Reader, src fs.ObjectInfo, op
 	var resp *http.Response
 	var info api.UploadSpecification
 	opts := rest.Opts{
-		Method: "POST",
-		Path:   "/Items(" + directoryID + ")/Upload2",
+		Method:  "POST",
+		Path:    "/Items(" + directoryID + ")/Upload2",
+		Options: options,
 	}
 	err = o.fs.pacer.Call(func() (bool, error) {
 		resp, err = o.fs.srv.CallJSON(ctx, &opts, &req, &info)
diff --git a/backend/sugarsync/sugarsync.go b/backend/sugarsync/sugarsync.go
index 2df0b0087..ae306caae 100644
--- a/backend/sugarsync/sugarsync.go
+++ b/backend/sugarsync/sugarsync.go
@@ -733,7 +733,7 @@ func (f *Fs) Put(ctx context.Context, in io.Reader, src fs.ObjectInfo, options .
 		return existingObj, existingObj.Update(ctx, in, src, options...)
 	case fs.ErrorObjectNotFound:
 		// Not found so create it
-		return f.PutUnchecked(ctx, in, src)
+		return f.PutUnchecked(ctx, in, src, options...)
 	default:
 		return nil, err
 	}
@@ -1320,6 +1320,7 @@ func (o *Object) Update(ctx context.Context, in io.Reader, src fs.ObjectInfo, op
 		RootURL:    o.id,
 		Path:       "/data",
 		NoResponse: true,
+		Options:    options,
 		Body:       in,
 	}
 	if size >= 0 {
diff --git a/backend/webdav/webdav.go b/backend/webdav/webdav.go
index 83fe1500f..3302df53a 100644
--- a/backend/webdav/webdav.go
+++ b/backend/webdav/webdav.go
@@ -1135,6 +1135,7 @@ func (o *Object) Update(ctx context.Context, in io.Reader, src fs.ObjectInfo, op
 		NoResponse:    true,
 		ContentLength: &size, // FIXME this isn't necessary with owncloud - See https://github.com/nextcloud/nextcloud-snap/issues/365
 		ContentType:   fs.MimeType(ctx, src),
+		Options:       options,
 	}
 	if o.fs.useOCMtime || o.fs.hasMD5 || o.fs.hasSHA1 {
 		opts.ExtraHeaders = map[string]string{}
diff --git a/backend/yandex/yandex.go b/backend/yandex/yandex.go
index 4fb9eb0b2..af134daa6 100644
--- a/backend/yandex/yandex.go
+++ b/backend/yandex/yandex.go
@@ -1065,7 +1065,7 @@ func (o *Object) Open(ctx context.Context, options ...fs.OpenOption) (in io.Read
 	return resp.Body, err
 }
 
-func (o *Object) upload(ctx context.Context, in io.Reader, overwrite bool, mimeType string) (err error) {
+func (o *Object) upload(ctx context.Context, in io.Reader, overwrite bool, mimeType string, options ...fs.OpenOption) (err error) {
 	// prepare upload
 	var resp *http.Response
 	var ur api.AsyncInfo
@@ -1073,6 +1073,7 @@ func (o *Object) upload(ctx context.Context, in io.Reader, overwrite bool, mimeT
 		Method:     "GET",
 		Path:       "/resources/upload",
 		Parameters: url.Values{},
+		Options:    options,
 	}
 
 	opts.Parameters.Set("path", o.fs.opt.Enc.FromStandardPath(o.filePath()))
@@ -1121,7 +1122,7 @@ func (o *Object) Update(ctx context.Context, in io.Reader, src fs.ObjectInfo, op
 	}
 
 	//upload file
-	err = o.upload(ctx, in1, true, fs.MimeType(ctx, src))
+	err = o.upload(ctx, in1, true, fs.MimeType(ctx, src), options...)
 	if err != nil {
 		return err
 	}
diff --git a/docs/content/docs.md b/docs/content/docs.md
index f559fbb47..838141f65 100644
--- a/docs/content/docs.md
+++ b/docs/content/docs.md
@@ -564,6 +564,29 @@ triggering follow-on actions if data was copied, or skipping if not.
 NB: Enabling this option turns a usually non-fatal error into a potentially
 fatal one - please check and adjust your scripts accordingly!
 
+### --header-download ###
+
+Add an HTTP header for all download transactions. The flag can be repeated to
+add multiple headers.
+
+```
+rclone sync s3:test/src ~/dst --header-download "X-Amz-Meta-Test: Foo" --header-download "X-Amz-Meta-Test2: Bar"
+```
+
+See the Github issue [here](https://github.com/rclone/rclone/issues/59) for
+currently supported backends.
+
+### --header-upload ###
+
+Add an HTTP header for all upload transactions. The flag can be repeated to add
+multiple headers.
+
+```
+rclone sync ~/src s3:test/dst --header-upload "Content-Disposition: attachment; filename='cool.html'" --header-upload "X-Amz-Meta-Test: FooBar"
+```
+
+See the Github issue [here](https://github.com/rclone/rclone/issues/59) for
+currently supported backends.
 
 ### --ignore-case-sync ###
 
diff --git a/docs/content/flags.md b/docs/content/flags.md
index 47001efa3..647829fcc 100755
--- a/docs/content/flags.md
+++ b/docs/content/flags.md
@@ -48,6 +48,8 @@ These flags are available for every command.
       --files-from stringArray               Read list of source-file names from file (use - to read from stdin).
   -f, --filter stringArray                   Add a file-filtering rule
       --filter-from stringArray              Read filtering patterns from a file (use - to read from stdin).
+      --header-download stringArray          Add HTTP header for download transactions
+      --header-upload stringArray            Add HTTP header for upload transactions
       --ignore-case                          Ignore case in filters (case insensitive)
       --ignore-case-sync                     Ignore case when synchronizing
       --ignore-checksum                      Skip post copy check of checksums.
diff --git a/fs/config.go b/fs/config.go
index 4e3cc6a60..555085c34 100644
--- a/fs/config.go
+++ b/fs/config.go
@@ -111,6 +111,8 @@ type ConfigInfo struct {
 	MultiThreadStreams     int
 	MultiThreadSet         bool   // whether MultiThreadStreams was set (set in fs/config/configflags)
 	OrderBy                string // instructions on how to order the transfer
+	UploadHeaders          []*HTTPOption
+	DownloadHeaders        []*HTTPOption
 }
 
 // NewConfig creates a new config with everything set to the default
diff --git a/fs/config/configflags/configflags.go b/fs/config/configflags/configflags.go
index d142f42c6..f06f83ec8 100644
--- a/fs/config/configflags/configflags.go
+++ b/fs/config/configflags/configflags.go
@@ -29,6 +29,8 @@ var (
 	deleteAfter     bool
 	bindAddr        string
 	disableFeatures string
+	uploadHeaders   []string
+	downloadHeaders []string
 )
 
 // AddFlags adds the non filing system specific flags to the command
@@ -112,6 +114,25 @@ func AddFlags(flagSet *pflag.FlagSet) {
 	flags.IntVarP(flagSet, &fs.Config.MultiThreadStreams, "multi-thread-streams", "", fs.Config.MultiThreadStreams, "Max number of streams to use for multi-thread downloads.")
 	flags.BoolVarP(flagSet, &fs.Config.UseJSONLog, "use-json-log", "", fs.Config.UseJSONLog, "Use json log format.")
 	flags.StringVarP(flagSet, &fs.Config.OrderBy, "order-by", "", fs.Config.OrderBy, "Instructions on how to order the transfers, eg 'size,descending'")
+	flags.StringArrayVarP(flagSet, &uploadHeaders, "header-upload", "", nil, "Set HTTP header for upload transactions")
+	flags.StringArrayVarP(flagSet, &downloadHeaders, "header-download", "", nil, "Set HTTP header for download transactions")
+}
+
+// ParseHeaders converts the strings passed in via the header flags into HTTPOptions
+func ParseHeaders(headers []string) []*fs.HTTPOption {
+	opts := []*fs.HTTPOption{}
+	for _, header := range headers {
+		parts := strings.SplitN(header, ":", 2)
+		if len(parts) == 1 {
+			log.Fatalf("Failed to parse '%s' as an HTTP header. Expecting a string like: 'Content-Encoding: gzip'", header)
+		}
+		option := &fs.HTTPOption{
+			Key:   strings.TrimSpace(parts[0]),
+			Value: strings.TrimSpace(parts[1]),
+		}
+		opts = append(opts, option)
+	}
+	return opts
 }
 
 // SetFlags converts any flags into config which weren't straight forward
@@ -212,6 +233,14 @@ func SetFlags() {
 		fs.Config.DisableFeatures = strings.Split(disableFeatures, ",")
 	}
 
+	if len(uploadHeaders) != 0 {
+		fs.Config.UploadHeaders = ParseHeaders(uploadHeaders)
+	}
+
+	if len(downloadHeaders) != 0 {
+		fs.Config.DownloadHeaders = ParseHeaders(downloadHeaders)
+	}
+
 	// Make the config file absolute
 	configPath, err := filepath.Abs(config.ConfigPath)
 	if err == nil {
diff --git a/fs/operations/multithread.go b/fs/operations/multithread.go
index 54c228583..c5bed49bf 100644
--- a/fs/operations/multithread.go
+++ b/fs/operations/multithread.go
@@ -71,7 +71,7 @@ func (mc *multiThreadCopyState) copyStream(ctx context.Context, stream int) (err
 
 	fs.Debugf(mc.src, "multi-thread copy: stream %d/%d (%d-%d) size %v starting", stream+1, mc.streams, start, end, fs.SizeSuffix(end-start))
 
-	rc, err := NewReOpen(ctx, mc.src, nil, &fs.RangeOption{Start: start, End: end - 1}, fs.Config.LowLevelRetries)
+	rc, err := NewReOpen(ctx, mc.src, fs.Config.LowLevelRetries, &fs.RangeOption{Start: start, End: end - 1})
 	if err != nil {
 		return errors.Wrap(err, "multpart copy: failed to open source")
 	}
diff --git a/fs/operations/operations.go b/fs/operations/operations.go
index 2d3a9046e..4d525459d 100644
--- a/fs/operations/operations.go
+++ b/fs/operations/operations.go
@@ -403,7 +403,11 @@ func Copy(ctx context.Context, f fs.Fs, dst fs.Object, remote string, src fs.Obj
 				}
 			} else {
 				var in0 io.ReadCloser
-				in0, err = NewReOpen(ctx, src, hashOption, nil, fs.Config.LowLevelRetries)
+				options := []fs.OpenOption{hashOption}
+				for _, option := range fs.Config.DownloadHeaders {
+					options = append(options, option)
+				}
+				in0, err = NewReOpen(ctx, src, fs.Config.LowLevelRetries, options...)
 				if err != nil {
 					err = errors.Wrap(err, "failed to open source object")
 				} else {
@@ -424,12 +428,16 @@ func Copy(ctx context.Context, f fs.Fs, dst fs.Object, remote string, src fs.Obj
 						if src.Remote() != remote {
 							wrappedSrc = NewOverrideRemote(src, remote)
 						}
+						options := []fs.OpenOption{hashOption}
+						for _, option := range fs.Config.UploadHeaders {
+							options = append(options, option)
+						}
 						if doUpdate {
 							actionTaken = "Copied (replaced existing)"
-							err = dst.Update(ctx, in, wrappedSrc, hashOption)
+							err = dst.Update(ctx, in, wrappedSrc, options...)
 						} else {
 							actionTaken = "Copied (new)"
-							dst, err = f.Put(ctx, in, wrappedSrc, hashOption)
+							dst, err = f.Put(ctx, in, wrappedSrc, options...)
 						}
 						closeErr := in.Close()
 						if err == nil {
diff --git a/fs/operations/reopen.go b/fs/operations/reopen.go
index bd45f96a5..688c5d057 100644
--- a/fs/operations/reopen.go
+++ b/fs/operations/reopen.go
@@ -12,17 +12,16 @@ import (
 
 // ReOpen is a wrapper for an object reader which reopens the stream on error
 type ReOpen struct {
-	ctx         context.Context
-	mu          sync.Mutex       // mutex to protect the below
-	src         fs.Object        // object to open
-	hashOption  *fs.HashesOption // option to pass to initial open
-	rangeOption *fs.RangeOption  // option to pass to initial open
-	rc          io.ReadCloser    // underlying stream
-	read        int64            // number of bytes read from this stream
-	maxTries    int              // maximum number of retries
-	tries       int              // number of retries we've had so far in this stream
-	err         error            // if this is set then Read/Close calls will return it
-	opened      bool             // if set then rc is valid and needs closing
+	ctx      context.Context
+	mu       sync.Mutex      // mutex to protect the below
+	src      fs.Object       // object to open
+	options  []fs.OpenOption // option to pass to initial open
+	rc       io.ReadCloser   // underlying stream
+	read     int64           // number of bytes read from this stream
+	maxTries int             // maximum number of retries
+	tries    int             // number of retries we've had so far in this stream
+	err      error           // if this is set then Read/Close calls will return it
+	opened   bool            // if set then rc is valid and needs closing
 }
 
 var (
@@ -36,13 +35,12 @@ var (
 //
 // If rangeOption is set then this will applied when reading from the
 // start, and updated on retries.
-func NewReOpen(ctx context.Context, src fs.Object, hashOption *fs.HashesOption, rangeOption *fs.RangeOption, maxTries int) (rc io.ReadCloser, err error) {
+func NewReOpen(ctx context.Context, src fs.Object, maxTries int, options ...fs.OpenOption) (rc io.ReadCloser, err error) {
 	h := &ReOpen{
-		ctx:         ctx,
-		src:         src,
-		hashOption:  hashOption,
-		rangeOption: rangeOption,
-		maxTries:    maxTries,
+		ctx:      ctx,
+		src:      src,
+		maxTries: maxTries,
+		options:  options,
 	}
 	h.mu.Lock()
 	defer h.mu.Unlock()
@@ -57,20 +55,35 @@ func NewReOpen(ctx context.Context, src fs.Object, hashOption *fs.HashesOption,
 //
 // we don't retry here as the Open() call will itself have low level retries
 func (h *ReOpen) open() error {
-	var optsArray [2]fs.OpenOption
-	var opts = optsArray[:0]
+	opts := []fs.OpenOption{}
+	var hashOption *fs.HashesOption
+	var rangeOption *fs.RangeOption
+	for _, option := range h.options {
+		switch option.(type) {
+		case *fs.HashesOption:
+			hashOption = option.(*fs.HashesOption)
+		case *fs.RangeOption:
+			rangeOption = option.(*fs.RangeOption)
+		case *fs.HTTPOption:
+			opts = append(opts, option)
+		default:
+			if option.Mandatory() {
+				fs.Logf(h.src, "Unsupported mandatory option: %v", option)
+			}
+		}
+	}
 	if h.read == 0 {
-		if h.rangeOption != nil {
-			opts = append(opts, h.rangeOption)
+		if rangeOption != nil {
+			opts = append(opts, rangeOption)
 		}
-		if h.hashOption != nil {
+		if hashOption != nil {
 			// put hashOption on if reading from the start, ditch otherwise
-			opts = append(opts, h.hashOption)
+			opts = append(opts, hashOption)
 		}
 	} else {
-		if h.rangeOption != nil {
+		if rangeOption != nil {
 			// range to the read point
-			opts = append(opts, &fs.RangeOption{Start: h.rangeOption.Start + h.read, End: h.rangeOption.End})
+			opts = append(opts, &fs.RangeOption{Start: rangeOption.Start + h.read, End: rangeOption.End})
 		} else {
 			// seek to the read point
 			opts = append(opts, &fs.SeekOption{Offset: h.read})


Test patch:
diff --git a/fs/operations/reopen_test.go b/fs/operations/reopen_test.go
index dac33e3cd..0faecb812 100644
--- a/fs/operations/reopen_test.go
+++ b/fs/operations/reopen_test.go
@@ -74,7 +74,7 @@ func TestReOpen(t *testing.T) {
 					breaks: breaks,
 				}
 				hashOption := &fs.HashesOption{Hashes: hash.NewHashSet(hash.MD5)}
-				return NewReOpen(context.Background(), src, hashOption, rangeOption, maxRetries)
+				return NewReOpen(context.Background(), src, maxRetries, hashOption, rangeOption)
 			}
 
 			t.Run("Basics", func(t *testing.T) {