seaweedfs/weed/s3api/s3api_object_copy_handlers.go

312 lines
9 KiB
Go
Raw Normal View History

package s3api
import (
"fmt"
"github.com/seaweedfs/seaweedfs/weed/glog"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
"modernc.org/strutil"
"net/http"
"net/url"
"strconv"
"strings"
"time"
"github.com/seaweedfs/seaweedfs/weed/util"
)
const (
DirectiveCopy = "COPY"
DirectiveReplace = "REPLACE"
)
func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request) {
dstBucket, dstObject := s3_constants.GetBucketAndObject(r)
// Copy source path.
cpSrcPath, err := url.QueryUnescape(r.Header.Get("X-Amz-Copy-Source"))
if err != nil {
// Save unescaped string as is.
cpSrcPath = r.Header.Get("X-Amz-Copy-Source")
}
srcBucket, srcObject := pathToBucketAndObject(cpSrcPath)
2021-09-19 07:18:59 +00:00
glog.V(3).Infof("CopyObjectHandler %s %s => %s %s", srcBucket, srcObject, dstBucket, dstObject)
replaceMeta, replaceTagging := replaceDirective(r.Header)
if (srcBucket == dstBucket && srcObject == dstObject || cpSrcPath == "") && (replaceMeta || replaceTagging) {
fullPath := util.FullPath(fmt.Sprintf("%s/%s%s", s3a.option.BucketsPath, dstBucket, dstObject))
dir, name := fullPath.DirAndName()
entry, err := s3a.getEntry(dir, name)
if err != nil || entry.IsDirectory {
s3err.WriteErrorResponse(w, r, s3err.ErrInvalidCopySource)
return
}
entry.Extended, err = processMetadataBytes(r.Header, entry.Extended, replaceMeta, replaceTagging)
entry.Attributes.Mtime = time.Now().Unix()
if err != nil {
glog.Errorf("CopyObjectHandler ValidateTags error %s: %v", r.URL, err)
s3err.WriteErrorResponse(w, r, s3err.ErrInvalidTag)
2022-06-28 15:24:03 +00:00
return
}
err = s3a.touch(dir, name, entry)
if err != nil {
2021-11-01 01:05:34 +00:00
s3err.WriteErrorResponse(w, r, s3err.ErrInvalidCopySource)
return
}
2021-11-01 01:02:08 +00:00
writeSuccessResponseXML(w, r, CopyObjectResult{
ETag: fmt.Sprintf("%x", entry.Attributes.Md5),
LastModified: time.Now().UTC(),
2021-06-11 04:50:21 +00:00
})
return
}
// If source object is empty or bucket is empty, reply back invalid copy source.
if srcObject == "" || srcBucket == "" {
2021-11-01 01:05:34 +00:00
s3err.WriteErrorResponse(w, r, s3err.ErrInvalidCopySource)
return
}
2021-09-06 06:37:40 +00:00
srcPath := util.FullPath(fmt.Sprintf("%s/%s%s", s3a.option.BucketsPath, srcBucket, srcObject))
dir, name := srcPath.DirAndName()
if entry, err := s3a.getEntry(dir, name); err != nil || entry.IsDirectory {
2021-11-01 01:05:34 +00:00
s3err.WriteErrorResponse(w, r, s3err.ErrInvalidCopySource)
return
}
if srcBucket == dstBucket && srcObject == dstObject {
2021-11-01 01:05:34 +00:00
s3err.WriteErrorResponse(w, r, s3err.ErrInvalidCopyDest)
return
}
dstUrl := fmt.Sprintf("http://%s%s/%s%s",
s3a.option.Filer.ToHttpAddress(), s3a.option.BucketsPath, dstBucket, urlEscapeObject(dstObject))
srcUrl := fmt.Sprintf("http://%s%s/%s%s",
s3a.option.Filer.ToHttpAddress(), s3a.option.BucketsPath, srcBucket, urlEscapeObject(srcObject))
FEATURE: add JWT to HTTP endpoints of Filer and use them in S3 Client - one JWT for reading and one for writing, analogous to how the JWT between Master and Volume Server works - I did not implement IP `whiteList` parameter on the filer Additionally, because http_util.DownloadFile now sets the JWT, the `download` command should now work when `jwt.signing.read` is configured. By looking at the code, I think this case did not work before. ## Docs to be adjusted after a release Page `Amazon-S3-API`: ``` # Authentication with Filer You can use mTLS for the gRPC connection between S3-API-Proxy and the filer, as explained in [Security-Configuration](Security-Configuration) - controlled by the `grpc.*` configuration in `security.toml`. Starting with version XX, it is also possible to authenticate the HTTP operations between the S3-API-Proxy and the Filer (especially uploading new files). This is configured by setting `filer_jwt.signing.key` and `filer_jwt.signing.read.key` in `security.toml`. With both configurations (gRPC and JWT), it is possible to have Filer and S3 communicate in fully authenticated fashion; so Filer will reject any unauthenticated communication. ``` Page `Security Overview`: ``` The following items are not covered, yet: - master server http REST services Starting with version XX, the Filer HTTP REST services can be secured with a JWT, by setting `filer_jwt.signing.key` and `filer_jwt.signing.read.key` in `security.toml`. ... Before version XX: "weed filer -disableHttp", disable http operations, only gRPC operations are allowed. This works with "weed mount" by FUSE. It does **not work** with the [S3 Gateway](Amazon S3 API), as this does HTTP calls to the Filer. Starting with version XX: secured by JWT, by setting `filer_jwt.signing.key` and `filer_jwt.signing.read.key` in `security.toml`. **This now works with the [S3 Gateway](Amazon S3 API).** ... # Securing Filer HTTP with JWT To enable JWT-based access control for the Filer, 1. generate `security.toml` file by `weed scaffold -config=security` 2. set `filer_jwt.signing.key` to a secret string - and optionally filer_jwt.signing.read.key` as well to a secret string 3. copy the same `security.toml` file to the filers and all S3 proxies. If `filer_jwt.signing.key` is configured: When sending upload/update/delete HTTP operations to a filer server, the request header `Authorization` should be the JWT string (`Authorization: Bearer [JwtToken]`). The operation is authorized after the filer validates the JWT with `filer_jwt.signing.key`. If `filer_jwt.signing.read.key` is configured: When sending GET or HEAD requests to a filer server, the request header `Authorization` should be the JWT string (`Authorization: Bearer [JwtToken]`). The operation is authorized after the filer validates the JWT with `filer_jwt.signing.read.key`. The S3 API Gateway reads the above JWT keys and sends authenticated HTTP requests to the filer. ``` Page `Security Configuration`: ``` (update scaffold file) ... [filer_jwt.signing] key = "blahblahblahblah" [filer_jwt.signing.read] key = "blahblahblahblah" ``` Resolves: #158
2021-12-29 18:47:53 +00:00
_, _, resp, err := util.DownloadFile(srcUrl, s3a.maybeGetFilerJwtAuthorizationToken(false))
if err != nil {
2021-11-01 01:05:34 +00:00
s3err.WriteErrorResponse(w, r, s3err.ErrInvalidCopySource)
return
}
2020-09-09 10:53:09 +00:00
defer util.CloseResponse(resp)
tagErr := processMetadata(r.Header, resp.Header, replaceMeta, replaceTagging, s3a.getTags, dir, name)
if tagErr != nil {
s3err.WriteErrorResponse(w, r, s3err.ErrInvalidCopySource)
return
}
2020-11-11 10:01:24 +00:00
glog.V(2).Infof("copy from %s to %s", srcUrl, dstUrl)
destination := fmt.Sprintf("%s/%s%s", s3a.option.BucketsPath, dstBucket, dstObject)
etag, errCode := s3a.putToFiler(r, dstUrl, resp.Body, destination, dstBucket)
2020-09-19 21:09:58 +00:00
if errCode != s3err.ErrNone {
2021-11-01 01:05:34 +00:00
s3err.WriteErrorResponse(w, r, errCode)
return
}
setEtag(w, etag)
response := CopyObjectResult{
ETag: etag,
LastModified: time.Now().UTC(),
}
2021-11-01 01:02:08 +00:00
writeSuccessResponseXML(w, r, response)
}
func pathToBucketAndObject(path string) (bucket, object string) {
path = strings.TrimPrefix(path, "/")
parts := strings.SplitN(path, "/", 2)
if len(parts) == 2 {
return parts[0], "/" + parts[1]
}
return parts[0], "/"
}
type CopyPartResult struct {
LastModified time.Time `xml:"LastModified"`
ETag string `xml:"ETag"`
}
func (s3a *S3ApiServer) CopyObjectPartHandler(w http.ResponseWriter, r *http.Request) {
// https://docs.aws.amazon.com/AmazonS3/latest/dev/CopyingObjctsUsingRESTMPUapi.html
// https://docs.aws.amazon.com/AmazonS3/latest/API/API_UploadPartCopy.html
dstBucket, dstObject := s3_constants.GetBucketAndObject(r)
// Copy source path.
cpSrcPath, err := url.QueryUnescape(r.Header.Get("X-Amz-Copy-Source"))
if err != nil {
// Save unescaped string as is.
cpSrcPath = r.Header.Get("X-Amz-Copy-Source")
}
srcBucket, srcObject := pathToBucketAndObject(cpSrcPath)
// If source object is empty or bucket is empty, reply back invalid copy source.
if srcObject == "" || srcBucket == "" {
2021-11-01 01:05:34 +00:00
s3err.WriteErrorResponse(w, r, s3err.ErrInvalidCopySource)
return
}
uploadID := r.URL.Query().Get("uploadId")
partIDString := r.URL.Query().Get("partNumber")
partID, err := strconv.Atoi(partIDString)
if err != nil {
2021-11-01 01:05:34 +00:00
s3err.WriteErrorResponse(w, r, s3err.ErrInvalidPart)
return
}
2021-09-19 07:18:59 +00:00
glog.V(3).Infof("CopyObjectPartHandler %s %s => %s part %d", srcBucket, srcObject, dstBucket, partID)
// check partID with maximum part ID for multipart objects
2020-02-25 20:58:45 +00:00
if partID > globalMaxPartID {
2021-11-01 01:05:34 +00:00
s3err.WriteErrorResponse(w, r, s3err.ErrInvalidMaxParts)
return
}
rangeHeader := r.Header.Get("x-amz-copy-source-range")
dstUrl := fmt.Sprintf("http://%s%s/%s/%04d.part",
s3a.option.Filer.ToHttpAddress(), s3a.genUploadsFolder(dstBucket), uploadID, partID)
srcUrl := fmt.Sprintf("http://%s%s/%s%s",
s3a.option.Filer.ToHttpAddress(), s3a.option.BucketsPath, srcBucket, urlEscapeObject(srcObject))
resp, dataReader, err := util.ReadUrlAsReaderCloser(srcUrl, s3a.maybeGetFilerJwtAuthorizationToken(false), rangeHeader)
if err != nil {
2021-11-01 01:05:34 +00:00
s3err.WriteErrorResponse(w, r, s3err.ErrInvalidCopySource)
return
}
defer util.CloseResponse(resp)
2020-02-14 17:09:15 +00:00
defer dataReader.Close()
2020-11-11 10:01:24 +00:00
glog.V(2).Infof("copy from %s to %s", srcUrl, dstUrl)
destination := fmt.Sprintf("%s/%s%s", s3a.option.BucketsPath, dstBucket, dstObject)
etag, errCode := s3a.putToFiler(r, dstUrl, dataReader, destination, dstBucket)
2020-09-19 21:09:58 +00:00
if errCode != s3err.ErrNone {
2021-11-01 01:05:34 +00:00
s3err.WriteErrorResponse(w, r, errCode)
return
}
setEtag(w, etag)
response := CopyPartResult{
ETag: etag,
LastModified: time.Now().UTC(),
}
2021-11-01 01:02:08 +00:00
writeSuccessResponseXML(w, r, response)
}
func replaceDirective(reqHeader http.Header) (replaceMeta, replaceTagging bool) {
return reqHeader.Get(s3_constants.AmzUserMetaDirective) == DirectiveReplace, reqHeader.Get(s3_constants.AmzObjectTaggingDirective) == DirectiveReplace
}
func processMetadata(reqHeader, existing http.Header, replaceMeta, replaceTagging bool, getTags func(parentDirectoryPath string, entryName string) (tags map[string]string, err error), dir, name string) (err error) {
if sc := reqHeader.Get(s3_constants.AmzStorageClass); len(sc) == 0 {
if sc := existing[s3_constants.AmzStorageClass]; len(sc) > 0 {
reqHeader[s3_constants.AmzStorageClass] = sc
}
}
if !replaceMeta {
for header, _ := range reqHeader {
if strings.HasPrefix(header, s3_constants.AmzUserMetaPrefix) {
delete(reqHeader, header)
}
}
for k, v := range existing {
if strings.HasPrefix(k, s3_constants.AmzUserMetaPrefix) {
reqHeader[k] = v
}
}
}
if !replaceTagging {
for header, _ := range reqHeader {
if strings.HasPrefix(header, s3_constants.AmzObjectTagging) {
delete(reqHeader, header)
}
}
found := false
for k, _ := range existing {
if strings.HasPrefix(k, s3_constants.AmzObjectTaggingPrefix) {
found = true
break
}
}
if found {
tags, err := getTags(dir, name)
if err != nil {
return err
}
var tagArr []string
for k, v := range tags {
tagArr = append(tagArr, fmt.Sprintf("%s=%s", k, v))
}
tagStr := strutil.JoinFields(tagArr, "&")
reqHeader.Set(s3_constants.AmzObjectTagging, tagStr)
}
}
return
}
func processMetadataBytes(reqHeader http.Header, existing map[string][]byte, replaceMeta, replaceTagging bool) (metadata map[string][]byte, err error) {
metadata = make(map[string][]byte)
if sc := existing[s3_constants.AmzStorageClass]; len(sc) > 0 {
metadata[s3_constants.AmzStorageClass] = sc
}
if sc := reqHeader.Get(s3_constants.AmzStorageClass); len(sc) > 0 {
metadata[s3_constants.AmzStorageClass] = []byte(sc)
}
if replaceMeta {
for header, values := range reqHeader {
if strings.HasPrefix(header, s3_constants.AmzUserMetaPrefix) {
for _, value := range values {
metadata[header] = []byte(value)
}
}
}
} else {
for k, v := range existing {
if strings.HasPrefix(k, s3_constants.AmzUserMetaPrefix) {
metadata[k] = v
}
}
}
if replaceTagging {
if tags := reqHeader.Get(s3_constants.AmzObjectTagging); tags != "" {
parsedTags, err := parseTagsHeader(tags)
if err != nil {
return nil, err
}
err = ValidateTags(parsedTags)
if err != nil {
return nil, err
}
for k, v := range parsedTags {
metadata[s3_constants.AmzObjectTagging+"-"+k] = []byte(v)
}
}
} else {
for k, v := range existing {
if strings.HasPrefix(k, s3_constants.AmzObjectTagging) {
metadata[k] = v
}
}
delete(metadata, s3_constants.AmzTagCount)
}
return
}