forked from TrueCloudLab/frostfs-s3-gw
744 lines
23 KiB
Go
744 lines
23 KiB
Go
package handler
|
|
|
|
import (
|
|
"encoding/xml"
|
|
"fmt"
|
|
"net/http"
|
|
"net/url"
|
|
"strconv"
|
|
"time"
|
|
|
|
"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api/data"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api/errors"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api/layer"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api/middleware"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/internal/logs"
|
|
"github.com/google/uuid"
|
|
"go.uber.org/zap"
|
|
)
|
|
|
|
type (
|
|
InitiateMultipartUploadResponse struct {
|
|
XMLName xml.Name `xml:"http://s3.amazonaws.com/doc/2006-03-01/ InitiateMultipartUploadResult" json:"-"`
|
|
Bucket string `xml:"Bucket"`
|
|
Key string `xml:"Key"`
|
|
UploadID string `xml:"UploadId"`
|
|
}
|
|
|
|
CompleteMultipartUploadResponse struct {
|
|
XMLName xml.Name `xml:"http://s3.amazonaws.com/doc/2006-03-01/ CompleteMultipartUploadResult" json:"-"`
|
|
Bucket string `xml:"Bucket"`
|
|
Key string `xml:"Key"`
|
|
ETag string `xml:"ETag"`
|
|
}
|
|
|
|
ListMultipartUploadsResponse struct {
|
|
XMLName xml.Name `xml:"http://s3.amazonaws.com/doc/2006-03-01/ ListMultipartUploadsResult" json:"-"`
|
|
Bucket string `xml:"Bucket"`
|
|
CommonPrefixes []CommonPrefix `xml:"CommonPrefixes"`
|
|
Delimiter string `xml:"Delimiter,omitempty"`
|
|
EncodingType string `xml:"EncodingType,omitempty"`
|
|
IsTruncated bool `xml:"IsTruncated"`
|
|
KeyMarker string `xml:"KeyMarker"`
|
|
MaxUploads int `xml:"MaxUploads"`
|
|
NextKeyMarker string `xml:"NextKeyMarker,omitempty"`
|
|
NextUploadIDMarker string `xml:"NextUploadIdMarker,omitempty"`
|
|
Prefix string `xml:"Prefix"`
|
|
Uploads []MultipartUpload `xml:"Upload"`
|
|
UploadIDMarker string `xml:"UploadIdMarker,omitempty"`
|
|
}
|
|
|
|
ListPartsResponse struct {
|
|
XMLName xml.Name `xml:"http://s3.amazonaws.com/doc/2006-03-01/ ListPartsResult" json:"-"`
|
|
Bucket string `xml:"Bucket"`
|
|
Initiator Initiator `xml:"Initiator"`
|
|
IsTruncated bool `xml:"IsTruncated"`
|
|
Key string `xml:"Key"`
|
|
MaxParts int `xml:"MaxParts,omitempty"`
|
|
NextPartNumberMarker int `xml:"NextPartNumberMarker,omitempty"`
|
|
Owner Owner `xml:"Owner"`
|
|
Parts []*layer.Part `xml:"Part"`
|
|
PartNumberMarker int `xml:"PartNumberMarker,omitempty"`
|
|
StorageClass string `xml:"StorageClass"`
|
|
UploadID string `xml:"UploadId"`
|
|
}
|
|
|
|
MultipartUpload struct {
|
|
Initiated string `xml:"Initiated"`
|
|
Initiator Initiator `xml:"Initiator"`
|
|
Key string `xml:"Key"`
|
|
Owner Owner `xml:"Owner"`
|
|
StorageClass string `xml:"StorageClass"`
|
|
UploadID string `xml:"UploadId"`
|
|
}
|
|
|
|
Initiator struct {
|
|
ID string `xml:"ID"`
|
|
DisplayName string `xml:"DisplayName"`
|
|
}
|
|
|
|
CompleteMultipartUpload struct {
|
|
XMLName xml.Name `xml:"http://s3.amazonaws.com/doc/2006-03-01/ CompleteMultipartUpload"`
|
|
Parts []*layer.CompletedPart `xml:"Part"`
|
|
}
|
|
|
|
UploadPartCopyResponse struct {
|
|
ETag string `xml:"ETag"`
|
|
LastModified string `xml:"LastModified"`
|
|
}
|
|
)
|
|
|
|
const (
|
|
uploadIDHeaderName = "uploadId"
|
|
partNumberHeaderName = "partNumber"
|
|
|
|
prefixQueryName = "prefix"
|
|
delimiterQueryName = "delimiter"
|
|
maxUploadsQueryName = "max-uploads"
|
|
encodingTypeQueryName = "encoding-type"
|
|
keyMarkerQueryName = "key-marker"
|
|
uploadIDMarkerQueryName = "upload-id-marker"
|
|
)
|
|
|
|
func (h *handler) CreateMultipartUploadHandler(w http.ResponseWriter, r *http.Request) {
|
|
reqInfo := middleware.GetReqInfo(r.Context())
|
|
uploadID := uuid.New()
|
|
cannedACLStatus := aclHeadersStatus(r)
|
|
additional := []zap.Field{zap.String("uploadID", uploadID.String())}
|
|
|
|
bktInfo, err := h.getBucketAndCheckOwner(r, reqInfo.BucketName)
|
|
if err != nil {
|
|
h.logAndSendError(w, "could not get bucket info", reqInfo, err)
|
|
return
|
|
}
|
|
|
|
settings, err := h.obj.GetBucketSettings(r.Context(), bktInfo)
|
|
if err != nil {
|
|
h.logAndSendError(w, "couldn't get bucket settings", reqInfo, err)
|
|
return
|
|
}
|
|
|
|
apeEnabled := bktInfo.APEEnabled || settings.CannedACL != ""
|
|
if apeEnabled && cannedACLStatus == aclStatusYes {
|
|
h.logAndSendError(w, "acl not supported for this bucket", reqInfo, errors.GetAPIError(errors.ErrAccessControlListNotSupported))
|
|
return
|
|
}
|
|
|
|
p := &layer.CreateMultipartParams{
|
|
Info: &layer.UploadInfoParams{
|
|
UploadID: uploadID.String(),
|
|
Bkt: bktInfo,
|
|
Key: reqInfo.ObjectName,
|
|
},
|
|
Data: &layer.UploadData{},
|
|
}
|
|
|
|
needUpdateEACLTable := !(apeEnabled || cannedACLStatus == aclStatusNo)
|
|
if needUpdateEACLTable {
|
|
key, err := h.bearerTokenIssuerKey(r.Context())
|
|
if err != nil {
|
|
h.logAndSendError(w, "couldn't get gate key", reqInfo, err, additional...)
|
|
return
|
|
}
|
|
if _, err = parseACLHeaders(r.Header, key); err != nil {
|
|
h.logAndSendError(w, "could not parse acl", reqInfo, err, additional...)
|
|
return
|
|
}
|
|
p.Data.ACLHeaders = formACLHeadersForMultipart(r.Header)
|
|
}
|
|
|
|
if len(r.Header.Get(api.AmzTagging)) > 0 {
|
|
p.Data.TagSet, err = parseTaggingHeader(r.Header)
|
|
if err != nil {
|
|
h.logAndSendError(w, "could not parse tagging", reqInfo, err, additional...)
|
|
return
|
|
}
|
|
}
|
|
|
|
p.Info.Encryption, err = formEncryptionParams(r)
|
|
if err != nil {
|
|
h.logAndSendError(w, "invalid sse headers", reqInfo, err, additional...)
|
|
return
|
|
}
|
|
|
|
p.Header = parseMetadata(r)
|
|
if contentType := r.Header.Get(api.ContentType); len(contentType) > 0 {
|
|
p.Header[api.ContentType] = contentType
|
|
}
|
|
if contentLanguage := r.Header.Get(api.ContentLanguage); len(contentLanguage) > 0 {
|
|
p.Header[api.ContentLanguage] = contentLanguage
|
|
}
|
|
|
|
p.CopiesNumbers, err = h.pickCopiesNumbers(p.Header, reqInfo.Namespace, bktInfo.LocationConstraint)
|
|
if err != nil {
|
|
h.logAndSendError(w, "invalid copies number", reqInfo, err, additional...)
|
|
return
|
|
}
|
|
|
|
if err = h.obj.CreateMultipartUpload(r.Context(), p); err != nil {
|
|
h.logAndSendError(w, "could create multipart upload", reqInfo, err, additional...)
|
|
return
|
|
}
|
|
|
|
if p.Info.Encryption.Enabled() {
|
|
addSSECHeaders(w.Header(), r.Header)
|
|
}
|
|
|
|
resp := InitiateMultipartUploadResponse{
|
|
Bucket: reqInfo.BucketName,
|
|
Key: reqInfo.ObjectName,
|
|
UploadID: uploadID.String(),
|
|
}
|
|
|
|
if err = middleware.EncodeToResponse(w, resp); err != nil {
|
|
h.logAndSendError(w, "could not encode InitiateMultipartUploadResponse to response", reqInfo, err, additional...)
|
|
return
|
|
}
|
|
}
|
|
|
|
func formACLHeadersForMultipart(header http.Header) map[string]string {
|
|
result := make(map[string]string)
|
|
|
|
if value := header.Get(api.AmzACL); value != "" {
|
|
result[api.AmzACL] = value
|
|
}
|
|
if value := header.Get(api.AmzGrantRead); value != "" {
|
|
result[api.AmzGrantRead] = value
|
|
}
|
|
if value := header.Get(api.AmzGrantFullControl); value != "" {
|
|
result[api.AmzGrantFullControl] = value
|
|
}
|
|
if value := header.Get(api.AmzGrantWrite); value != "" {
|
|
result[api.AmzGrantWrite] = value
|
|
}
|
|
|
|
return result
|
|
}
|
|
|
|
func (h *handler) UploadPartHandler(w http.ResponseWriter, r *http.Request) {
|
|
reqInfo := middleware.GetReqInfo(r.Context())
|
|
|
|
bktInfo, err := h.getBucketAndCheckOwner(r, reqInfo.BucketName)
|
|
if err != nil {
|
|
h.logAndSendError(w, "could not get bucket info", reqInfo, err)
|
|
return
|
|
}
|
|
|
|
var (
|
|
queryValues = r.URL.Query()
|
|
uploadID = queryValues.Get(uploadIDHeaderName)
|
|
partNumStr = queryValues.Get(partNumberHeaderName)
|
|
additional = []zap.Field{zap.String("uploadID", uploadID), zap.String("partNumber", partNumStr)}
|
|
)
|
|
|
|
partNumber, err := strconv.Atoi(partNumStr)
|
|
if err != nil || partNumber < layer.UploadMinPartNumber || partNumber > layer.UploadMaxPartNumber {
|
|
h.logAndSendError(w, "invalid part number", reqInfo, errors.GetAPIError(errors.ErrInvalidPartNumber), additional...)
|
|
return
|
|
}
|
|
|
|
body, err := h.getBodyReader(r)
|
|
if err != nil {
|
|
h.logAndSendError(w, "failed to get body reader", reqInfo, err, additional...)
|
|
return
|
|
}
|
|
|
|
var size uint64
|
|
if r.ContentLength > 0 {
|
|
size = uint64(r.ContentLength)
|
|
}
|
|
|
|
p := &layer.UploadPartParams{
|
|
Info: &layer.UploadInfoParams{
|
|
UploadID: uploadID,
|
|
Bkt: bktInfo,
|
|
Key: reqInfo.ObjectName,
|
|
},
|
|
PartNumber: partNumber,
|
|
Size: size,
|
|
Reader: body,
|
|
ContentMD5: r.Header.Get(api.ContentMD5),
|
|
ContentSHA256Hash: r.Header.Get(api.AmzContentSha256),
|
|
}
|
|
|
|
p.Info.Encryption, err = formEncryptionParams(r)
|
|
if err != nil {
|
|
h.logAndSendError(w, "invalid sse headers", reqInfo, err, additional...)
|
|
return
|
|
}
|
|
|
|
hash, err := h.obj.UploadPart(r.Context(), p)
|
|
if err != nil {
|
|
h.logAndSendError(w, "could not upload a part", reqInfo, err, additional...)
|
|
return
|
|
}
|
|
|
|
if p.Info.Encryption.Enabled() {
|
|
addSSECHeaders(w.Header(), r.Header)
|
|
}
|
|
|
|
w.Header().Set(api.ETag, data.Quote(hash))
|
|
if err = middleware.WriteSuccessResponseHeadersOnly(w); err != nil {
|
|
h.logAndSendError(w, "write response", reqInfo, err)
|
|
return
|
|
}
|
|
}
|
|
|
|
func (h *handler) UploadPartCopy(w http.ResponseWriter, r *http.Request) {
|
|
var (
|
|
versionID string
|
|
ctx = r.Context()
|
|
reqInfo = middleware.GetReqInfo(ctx)
|
|
queryValues = reqInfo.URL.Query()
|
|
uploadID = queryValues.Get(uploadIDHeaderName)
|
|
partNumStr = queryValues.Get(partNumberHeaderName)
|
|
additional = []zap.Field{zap.String("uploadID", uploadID), zap.String("partNumber", partNumStr)}
|
|
)
|
|
|
|
partNumber, err := strconv.Atoi(partNumStr)
|
|
if err != nil || partNumber < layer.UploadMinPartNumber || partNumber > layer.UploadMaxPartNumber {
|
|
h.logAndSendError(w, "invalid part number", reqInfo, errors.GetAPIError(errors.ErrInvalidPartNumber), additional...)
|
|
return
|
|
}
|
|
|
|
src := r.Header.Get(api.AmzCopySource)
|
|
if u, err := url.Parse(src); err == nil {
|
|
versionID = u.Query().Get(api.QueryVersionID)
|
|
src = u.Path
|
|
}
|
|
srcBucket, srcObject, err := path2BucketObject(src)
|
|
if err != nil {
|
|
h.logAndSendError(w, "invalid source copy", reqInfo, err, additional...)
|
|
return
|
|
}
|
|
|
|
srcRange, err := parseRange(r.Header.Get(api.AmzCopySourceRange))
|
|
if err != nil {
|
|
h.logAndSendError(w, "could not parse copy range", reqInfo,
|
|
errors.GetAPIError(errors.ErrInvalidCopyPartRange), additional...)
|
|
return
|
|
}
|
|
|
|
srcBktInfo, err := h.getBucketAndCheckOwner(r, srcBucket, api.AmzSourceExpectedBucketOwner)
|
|
if err != nil {
|
|
h.logAndSendError(w, "could not get source bucket info", reqInfo, err, additional...)
|
|
return
|
|
}
|
|
|
|
bktInfo, err := h.getBucketAndCheckOwner(r, reqInfo.BucketName)
|
|
if err != nil {
|
|
h.logAndSendError(w, "could not get target bucket info", reqInfo, err, additional...)
|
|
return
|
|
}
|
|
|
|
headPrm := &layer.HeadObjectParams{
|
|
BktInfo: srcBktInfo,
|
|
Object: srcObject,
|
|
VersionID: versionID,
|
|
}
|
|
|
|
srcInfo, err := h.obj.GetObjectInfo(ctx, headPrm)
|
|
if err != nil {
|
|
if errors.IsS3Error(err, errors.ErrNoSuchKey) && versionID != "" {
|
|
h.logAndSendError(w, "could not head source object version", reqInfo,
|
|
errors.GetAPIError(errors.ErrBadRequest), additional...)
|
|
return
|
|
}
|
|
h.logAndSendError(w, "could not head source object", reqInfo, err, additional...)
|
|
return
|
|
}
|
|
|
|
args, err := parseCopyObjectArgs(r.Header)
|
|
if err != nil {
|
|
h.logAndSendError(w, "could not parse copy object args", reqInfo,
|
|
errors.GetAPIError(errors.ErrInvalidCopyPartRange), additional...)
|
|
return
|
|
}
|
|
|
|
if err = checkPreconditions(srcInfo, args.Conditional, h.cfg.MD5Enabled()); err != nil {
|
|
h.logAndSendError(w, "precondition failed", reqInfo, errors.GetAPIError(errors.ErrPreconditionFailed),
|
|
additional...)
|
|
return
|
|
}
|
|
|
|
srcEncryptionParams, err := formCopySourceEncryptionParams(r)
|
|
if err != nil {
|
|
h.logAndSendError(w, "invalid sse headers", reqInfo, err)
|
|
return
|
|
}
|
|
|
|
if err = srcEncryptionParams.MatchObjectEncryption(layer.FormEncryptionInfo(srcInfo.Headers)); err != nil {
|
|
h.logAndSendError(w, "encryption doesn't match object", reqInfo, fmt.Errorf("%w: %s", errors.GetAPIError(errors.ErrBadRequest), err), additional...)
|
|
return
|
|
}
|
|
|
|
p := &layer.UploadCopyParams{
|
|
Versioned: headPrm.Versioned(),
|
|
Info: &layer.UploadInfoParams{
|
|
UploadID: uploadID,
|
|
Bkt: bktInfo,
|
|
Key: reqInfo.ObjectName,
|
|
},
|
|
SrcObjInfo: srcInfo,
|
|
SrcBktInfo: srcBktInfo,
|
|
SrcEncryption: srcEncryptionParams,
|
|
PartNumber: partNumber,
|
|
Range: srcRange,
|
|
}
|
|
|
|
p.Info.Encryption, err = formEncryptionParams(r)
|
|
if err != nil {
|
|
h.logAndSendError(w, "invalid sse headers", reqInfo, err, additional...)
|
|
return
|
|
}
|
|
|
|
info, err := h.obj.UploadPartCopy(ctx, p)
|
|
if err != nil {
|
|
h.logAndSendError(w, "could not upload part copy", reqInfo, err, additional...)
|
|
return
|
|
}
|
|
|
|
response := UploadPartCopyResponse{
|
|
LastModified: info.Created.UTC().Format(time.RFC3339),
|
|
ETag: data.Quote(info.ETag(h.cfg.MD5Enabled())),
|
|
}
|
|
|
|
if p.Info.Encryption.Enabled() {
|
|
addSSECHeaders(w.Header(), r.Header)
|
|
}
|
|
|
|
if err = middleware.EncodeToResponse(w, response); err != nil {
|
|
h.logAndSendError(w, "something went wrong", reqInfo, err, additional...)
|
|
}
|
|
}
|
|
|
|
func (h *handler) CompleteMultipartUploadHandler(w http.ResponseWriter, r *http.Request) {
|
|
reqInfo := middleware.GetReqInfo(r.Context())
|
|
|
|
bktInfo, err := h.getBucketAndCheckOwner(r, reqInfo.BucketName)
|
|
if err != nil {
|
|
h.logAndSendError(w, "could not get bucket info", reqInfo, err)
|
|
return
|
|
}
|
|
|
|
settings, err := h.obj.GetBucketSettings(r.Context(), bktInfo)
|
|
if err != nil {
|
|
h.logAndSendError(w, "could not get bucket settings", reqInfo, err)
|
|
return
|
|
}
|
|
|
|
var (
|
|
uploadID = r.URL.Query().Get(uploadIDHeaderName)
|
|
uploadInfo = &layer.UploadInfoParams{
|
|
UploadID: uploadID,
|
|
Bkt: bktInfo,
|
|
Key: reqInfo.ObjectName,
|
|
}
|
|
additional = []zap.Field{zap.String("uploadID", uploadID)}
|
|
)
|
|
|
|
reqBody := new(CompleteMultipartUpload)
|
|
if err = h.cfg.NewXMLDecoder(r.Body).Decode(reqBody); err != nil {
|
|
h.logAndSendError(w, "could not read complete multipart upload xml", reqInfo,
|
|
errors.GetAPIError(errors.ErrMalformedXML), additional...)
|
|
return
|
|
}
|
|
if len(reqBody.Parts) == 0 {
|
|
h.logAndSendError(w, "invalid xml with parts", reqInfo, errors.GetAPIError(errors.ErrMalformedXML), additional...)
|
|
return
|
|
}
|
|
|
|
c := &layer.CompleteMultipartParams{
|
|
Info: uploadInfo,
|
|
Parts: reqBody.Parts,
|
|
}
|
|
|
|
// Start complete multipart upload which may take some time to fetch object
|
|
// and re-upload it part by part.
|
|
objInfo, err := h.completeMultipartUpload(r, c, bktInfo, reqInfo)
|
|
|
|
if err != nil {
|
|
h.logAndSendError(w, "complete multipart error", reqInfo, err, additional...)
|
|
return
|
|
}
|
|
|
|
response := CompleteMultipartUploadResponse{
|
|
Bucket: objInfo.Bucket,
|
|
Key: objInfo.Name,
|
|
ETag: data.Quote(objInfo.ETag(h.cfg.MD5Enabled())),
|
|
}
|
|
|
|
if settings.VersioningEnabled() {
|
|
w.Header().Set(api.AmzVersionID, objInfo.VersionID())
|
|
}
|
|
|
|
if err = middleware.EncodeToResponse(w, response); err != nil {
|
|
h.logAndSendError(w, "something went wrong", reqInfo, err, additional...)
|
|
}
|
|
}
|
|
|
|
func (h *handler) completeMultipartUpload(r *http.Request, c *layer.CompleteMultipartParams, bktInfo *data.BucketInfo, reqInfo *middleware.ReqInfo) (*data.ObjectInfo, error) {
|
|
ctx := r.Context()
|
|
uploadData, extendedObjInfo, err := h.obj.CompleteMultipartUpload(ctx, c)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("could not complete multipart upload: %w", err)
|
|
}
|
|
objInfo := extendedObjInfo.ObjectInfo
|
|
|
|
if len(uploadData.TagSet) != 0 {
|
|
tagPrm := &layer.PutObjectTaggingParams{
|
|
ObjectVersion: &layer.ObjectVersion{
|
|
BktInfo: bktInfo,
|
|
ObjectName: objInfo.Name,
|
|
VersionID: objInfo.VersionID(),
|
|
},
|
|
TagSet: uploadData.TagSet,
|
|
NodeVersion: extendedObjInfo.NodeVersion,
|
|
}
|
|
if _, err = h.obj.PutObjectTagging(ctx, tagPrm); err != nil {
|
|
return nil, fmt.Errorf("could not put tagging file of completed multipart upload: %w", err)
|
|
}
|
|
}
|
|
|
|
if len(uploadData.ACLHeaders) != 0 {
|
|
sessionTokenSetEACL, err := getSessionTokenSetEACL(ctx)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("couldn't get eacl token: %w", err)
|
|
}
|
|
key, err := h.bearerTokenIssuerKey(ctx)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("couldn't get gate key: %w", err)
|
|
}
|
|
acl, err := parseACLHeaders(r.Header, key)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("could not parse acl: %w", err)
|
|
}
|
|
|
|
resInfo := &resourceInfo{
|
|
Bucket: objInfo.Bucket,
|
|
Object: objInfo.Name,
|
|
}
|
|
astObject, err := aclToAst(acl, resInfo)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("could not translate acl of completed multipart upload to ast: %w", err)
|
|
}
|
|
if _, err = h.updateBucketACL(r, astObject, bktInfo, sessionTokenSetEACL); err != nil {
|
|
return nil, fmt.Errorf("could not update bucket acl while completing multipart upload: %w", err)
|
|
}
|
|
}
|
|
|
|
s := &SendNotificationParams{
|
|
Event: EventObjectCreatedCompleteMultipartUpload,
|
|
NotificationInfo: data.NotificationInfoFromObject(objInfo, h.cfg.MD5Enabled()),
|
|
BktInfo: bktInfo,
|
|
ReqInfo: reqInfo,
|
|
}
|
|
if err = h.sendNotifications(ctx, s); err != nil {
|
|
h.reqLogger(ctx).Error(logs.CouldntSendNotification, zap.Error(err))
|
|
}
|
|
|
|
return objInfo, nil
|
|
}
|
|
|
|
func (h *handler) ListMultipartUploadsHandler(w http.ResponseWriter, r *http.Request) {
|
|
reqInfo := middleware.GetReqInfo(r.Context())
|
|
|
|
bktInfo, err := h.getBucketAndCheckOwner(r, reqInfo.BucketName)
|
|
if err != nil {
|
|
h.logAndSendError(w, "could not get bucket info", reqInfo, err)
|
|
return
|
|
}
|
|
|
|
var (
|
|
queryValues = reqInfo.URL.Query()
|
|
maxUploadsStr = queryValues.Get(maxUploadsQueryName)
|
|
maxUploads = layer.MaxSizeUploadsList
|
|
)
|
|
|
|
if maxUploadsStr != "" {
|
|
val, err := strconv.Atoi(maxUploadsStr)
|
|
if err != nil || val < 1 || val > 1000 {
|
|
h.logAndSendError(w, "invalid maxUploads", reqInfo, errors.GetAPIError(errors.ErrInvalidMaxUploads))
|
|
return
|
|
}
|
|
maxUploads = val
|
|
}
|
|
|
|
p := &layer.ListMultipartUploadsParams{
|
|
Bkt: bktInfo,
|
|
Delimiter: queryValues.Get(delimiterQueryName),
|
|
EncodingType: queryValues.Get(encodingTypeQueryName),
|
|
KeyMarker: queryValues.Get(keyMarkerQueryName),
|
|
MaxUploads: maxUploads,
|
|
Prefix: queryValues.Get(prefixQueryName),
|
|
UploadIDMarker: queryValues.Get(uploadIDMarkerQueryName),
|
|
}
|
|
|
|
list, err := h.obj.ListMultipartUploads(r.Context(), p)
|
|
if err != nil {
|
|
h.logAndSendError(w, "could not list multipart uploads", reqInfo, err)
|
|
return
|
|
}
|
|
|
|
if err = middleware.EncodeToResponse(w, encodeListMultipartUploadsToResponse(list, p)); err != nil {
|
|
h.logAndSendError(w, "something went wrong", reqInfo, err)
|
|
}
|
|
}
|
|
|
|
func (h *handler) ListPartsHandler(w http.ResponseWriter, r *http.Request) {
|
|
reqInfo := middleware.GetReqInfo(r.Context())
|
|
|
|
bktInfo, err := h.getBucketAndCheckOwner(r, reqInfo.BucketName)
|
|
if err != nil {
|
|
h.logAndSendError(w, "could not get bucket info", reqInfo, err)
|
|
return
|
|
}
|
|
|
|
var (
|
|
partNumberMarker int
|
|
|
|
queryValues = reqInfo.URL.Query()
|
|
uploadID = queryValues.Get(uploadIDHeaderName)
|
|
additional = []zap.Field{zap.String("uploadID", uploadID), zap.String("Key", reqInfo.ObjectName)}
|
|
maxParts = layer.MaxSizePartsList
|
|
)
|
|
|
|
if queryValues.Get("max-parts") != "" {
|
|
val, err := strconv.Atoi(queryValues.Get("max-parts"))
|
|
if err != nil || val < 0 {
|
|
h.logAndSendError(w, "invalid MaxParts", reqInfo, errors.GetAPIError(errors.ErrInvalidMaxParts), additional...)
|
|
return
|
|
}
|
|
if val < layer.MaxSizePartsList {
|
|
maxParts = val
|
|
}
|
|
}
|
|
|
|
if queryValues.Get("part-number-marker") != "" {
|
|
if partNumberMarker, err = strconv.Atoi(queryValues.Get("part-number-marker")); err != nil || partNumberMarker < 0 {
|
|
h.logAndSendError(w, "invalid PartNumberMarker", reqInfo, err, additional...)
|
|
return
|
|
}
|
|
}
|
|
|
|
p := &layer.ListPartsParams{
|
|
Info: &layer.UploadInfoParams{
|
|
UploadID: uploadID,
|
|
Bkt: bktInfo,
|
|
Key: reqInfo.ObjectName,
|
|
},
|
|
MaxParts: maxParts,
|
|
PartNumberMarker: partNumberMarker,
|
|
}
|
|
|
|
p.Info.Encryption, err = formEncryptionParams(r)
|
|
if err != nil {
|
|
h.logAndSendError(w, "invalid sse headers", reqInfo, err)
|
|
return
|
|
}
|
|
|
|
list, err := h.obj.ListParts(r.Context(), p)
|
|
if err != nil {
|
|
h.logAndSendError(w, "could not list parts", reqInfo, err, additional...)
|
|
return
|
|
}
|
|
|
|
if err = middleware.EncodeToResponse(w, encodeListPartsToResponse(list, p)); err != nil {
|
|
h.logAndSendError(w, "something went wrong", reqInfo, err)
|
|
}
|
|
}
|
|
|
|
func (h *handler) AbortMultipartUploadHandler(w http.ResponseWriter, r *http.Request) {
|
|
reqInfo := middleware.GetReqInfo(r.Context())
|
|
|
|
bktInfo, err := h.getBucketAndCheckOwner(r, reqInfo.BucketName)
|
|
if err != nil {
|
|
h.logAndSendError(w, "could not get bucket info", reqInfo, err)
|
|
return
|
|
}
|
|
|
|
uploadID := reqInfo.URL.Query().Get(uploadIDHeaderName)
|
|
additional := []zap.Field{zap.String("uploadID", uploadID), zap.String("Key", reqInfo.ObjectName)}
|
|
|
|
p := &layer.UploadInfoParams{
|
|
UploadID: uploadID,
|
|
Bkt: bktInfo,
|
|
Key: reqInfo.ObjectName,
|
|
}
|
|
|
|
p.Encryption, err = formEncryptionParams(r)
|
|
if err != nil {
|
|
h.logAndSendError(w, "invalid sse headers", reqInfo, err)
|
|
return
|
|
}
|
|
|
|
if err = h.obj.AbortMultipartUpload(r.Context(), p); err != nil {
|
|
h.logAndSendError(w, "could not abort multipart upload", reqInfo, err, additional...)
|
|
return
|
|
}
|
|
|
|
w.WriteHeader(http.StatusNoContent)
|
|
}
|
|
|
|
func encodeListMultipartUploadsToResponse(info *layer.ListMultipartUploadsInfo, params *layer.ListMultipartUploadsParams) *ListMultipartUploadsResponse {
|
|
res := ListMultipartUploadsResponse{
|
|
Bucket: params.Bkt.Name,
|
|
CommonPrefixes: fillPrefixes(info.Prefixes, params.EncodingType),
|
|
Delimiter: params.Delimiter,
|
|
EncodingType: params.EncodingType,
|
|
IsTruncated: info.IsTruncated,
|
|
KeyMarker: params.KeyMarker,
|
|
MaxUploads: params.MaxUploads,
|
|
NextKeyMarker: info.NextKeyMarker,
|
|
NextUploadIDMarker: info.NextUploadIDMarker,
|
|
Prefix: params.Prefix,
|
|
UploadIDMarker: params.UploadIDMarker,
|
|
}
|
|
|
|
uploads := make([]MultipartUpload, 0, len(info.Uploads))
|
|
for _, u := range info.Uploads {
|
|
m := MultipartUpload{
|
|
Initiated: u.Created.UTC().Format(time.RFC3339),
|
|
Initiator: Initiator{
|
|
ID: u.Owner.String(),
|
|
DisplayName: u.Owner.String(),
|
|
},
|
|
Key: u.Key,
|
|
Owner: Owner{
|
|
ID: u.Owner.String(),
|
|
DisplayName: u.Owner.String(),
|
|
},
|
|
UploadID: u.UploadID,
|
|
StorageClass: api.DefaultStorageClass,
|
|
}
|
|
uploads = append(uploads, m)
|
|
}
|
|
|
|
res.Uploads = uploads
|
|
|
|
return &res
|
|
}
|
|
|
|
func encodeListPartsToResponse(info *layer.ListPartsInfo, params *layer.ListPartsParams) *ListPartsResponse {
|
|
return &ListPartsResponse{
|
|
XMLName: xml.Name{},
|
|
Bucket: params.Info.Bkt.Name,
|
|
Initiator: Initiator{
|
|
ID: info.Owner.String(),
|
|
DisplayName: info.Owner.String(),
|
|
},
|
|
IsTruncated: info.IsTruncated,
|
|
Key: params.Info.Key,
|
|
MaxParts: params.MaxParts,
|
|
NextPartNumberMarker: info.NextPartNumberMarker,
|
|
Owner: Owner{
|
|
ID: info.Owner.String(),
|
|
DisplayName: info.Owner.String(),
|
|
},
|
|
PartNumberMarker: params.PartNumberMarker,
|
|
UploadID: params.Info.UploadID,
|
|
Parts: info.Parts,
|
|
StorageClass: api.DefaultStorageClass,
|
|
}
|
|
}
|