forked from TrueCloudLab/frostfs-s3-gw
[#205] Add md5 checksum in ETag by config param
Signed-off-by: Marina Biryukova <m.biryukova@yadro.com>
This commit is contained in:
parent
962c120125
commit
aa2c016f83
30 changed files with 313 additions and 82 deletions
|
@ -45,6 +45,7 @@ type (
|
|||
Created time.Time
|
||||
CreationEpoch uint64
|
||||
HashSum string
|
||||
MD5Sum string
|
||||
Owner user.ID
|
||||
Headers map[string]string
|
||||
}
|
||||
|
@ -81,12 +82,12 @@ type (
|
|||
)
|
||||
|
||||
// NotificationInfoFromObject creates new NotificationInfo from ObjectInfo.
|
||||
func NotificationInfoFromObject(objInfo *ObjectInfo) *NotificationInfo {
|
||||
func NotificationInfoFromObject(objInfo *ObjectInfo, md5Enabled bool) *NotificationInfo {
|
||||
return &NotificationInfo{
|
||||
Name: objInfo.Name,
|
||||
Version: objInfo.VersionID(),
|
||||
Size: objInfo.Size,
|
||||
HashSum: objInfo.HashSum,
|
||||
HashSum: objInfo.ETag(md5Enabled),
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -115,6 +116,13 @@ func (o *ObjectInfo) Address() oid.Address {
|
|||
return addr
|
||||
}
|
||||
|
||||
func (o *ObjectInfo) ETag(md5Enabled bool) string {
|
||||
if md5Enabled && len(o.MD5Sum) > 0 {
|
||||
return o.MD5Sum
|
||||
}
|
||||
return o.HashSum
|
||||
}
|
||||
|
||||
func (b BucketSettings) Unversioned() bool {
|
||||
return b.Versioning == VersioningUnversioned
|
||||
}
|
||||
|
|
|
@ -56,6 +56,7 @@ type BaseNodeVersion struct {
|
|||
Timestamp uint64
|
||||
Size uint64
|
||||
ETag string
|
||||
MD5 string
|
||||
FilePath string
|
||||
}
|
||||
|
||||
|
@ -86,14 +87,23 @@ type PartInfo struct {
|
|||
OID oid.ID `json:"oid"`
|
||||
Size uint64 `json:"size"`
|
||||
ETag string `json:"etag"`
|
||||
MD5 string `json:"md5"`
|
||||
Created time.Time `json:"created"`
|
||||
}
|
||||
|
||||
// ToHeaderString form short part representation to use in S3-Completed-Parts header.
|
||||
func (p *PartInfo) ToHeaderString() string {
|
||||
// ETag value contains SHA256 checksum which is used while getting object parts attributes.
|
||||
return strconv.Itoa(p.Number) + "-" + strconv.FormatUint(p.Size, 10) + "-" + p.ETag
|
||||
}
|
||||
|
||||
func (p *PartInfo) GetETag(md5Enabled bool) string {
|
||||
if md5Enabled && len(p.MD5) > 0 {
|
||||
return p.MD5
|
||||
}
|
||||
return p.ETag
|
||||
}
|
||||
|
||||
// LockInfo is lock information to create appropriate tree node.
|
||||
type LockInfo struct {
|
||||
id uint64
|
||||
|
|
|
@ -466,7 +466,7 @@ func (h *handler) PutObjectACLHandler(w http.ResponseWriter, r *http.Request) {
|
|||
if updated {
|
||||
s := &SendNotificationParams{
|
||||
Event: EventObjectACLPut,
|
||||
NotificationInfo: data.NotificationInfoFromObject(objInfo),
|
||||
NotificationInfo: data.NotificationInfoFromObject(objInfo, h.cfg.Features.MD5Enabled()),
|
||||
BktInfo: bktInfo,
|
||||
ReqInfo: reqInfo,
|
||||
}
|
||||
|
|
|
@ -39,6 +39,7 @@ type (
|
|||
IsResolveListAllow bool // True if ResolveZoneList contains allowed zones
|
||||
CompleteMultipartKeepalive time.Duration
|
||||
Kludge KludgeSettings
|
||||
Features layer.FeatureSettings
|
||||
}
|
||||
|
||||
PlacementPolicy interface {
|
||||
|
|
|
@ -1,6 +1,8 @@
|
|||
package handler
|
||||
|
||||
import (
|
||||
"encoding/base64"
|
||||
"encoding/hex"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"strconv"
|
||||
|
@ -106,7 +108,7 @@ func (h *handler) GetObjectAttributesHandler(w http.ResponseWriter, r *http.Requ
|
|||
return
|
||||
}
|
||||
|
||||
if err = checkPreconditions(info, params.Conditional); err != nil {
|
||||
if err = checkPreconditions(info, params.Conditional, h.cfg.Features.MD5Enabled()); err != nil {
|
||||
h.logAndSendError(w, "precondition failed", reqInfo, err)
|
||||
return
|
||||
}
|
||||
|
@ -117,7 +119,7 @@ func (h *handler) GetObjectAttributesHandler(w http.ResponseWriter, r *http.Requ
|
|||
return
|
||||
}
|
||||
|
||||
response, err := encodeToObjectAttributesResponse(info, params)
|
||||
response, err := encodeToObjectAttributesResponse(info, params, h.cfg.Features.MD5Enabled())
|
||||
if err != nil {
|
||||
h.logAndSendError(w, "couldn't encode object info to response", reqInfo, err)
|
||||
return
|
||||
|
@ -179,19 +181,23 @@ func parseGetObjectAttributeArgs(r *http.Request) (*GetObjectAttributesArgs, err
|
|||
return res, err
|
||||
}
|
||||
|
||||
func encodeToObjectAttributesResponse(info *data.ObjectInfo, p *GetObjectAttributesArgs) (*GetObjectAttributesResponse, error) {
|
||||
func encodeToObjectAttributesResponse(info *data.ObjectInfo, p *GetObjectAttributesArgs, md5Enabled bool) (*GetObjectAttributesResponse, error) {
|
||||
resp := &GetObjectAttributesResponse{}
|
||||
|
||||
for _, attr := range p.Attributes {
|
||||
switch attr {
|
||||
case eTag:
|
||||
resp.ETag = info.HashSum
|
||||
resp.ETag = info.ETag(md5Enabled)
|
||||
case storageClass:
|
||||
resp.StorageClass = "STANDARD"
|
||||
case objectSize:
|
||||
resp.ObjectSize = info.Size
|
||||
case checksum:
|
||||
resp.Checksum = &Checksum{ChecksumSHA256: info.HashSum}
|
||||
checksumBytes, err := hex.DecodeString(info.HashSum)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("form upload attributes: %w", err)
|
||||
}
|
||||
resp.Checksum = &Checksum{ChecksumSHA256: base64.StdEncoding.EncodeToString(checksumBytes)}
|
||||
case objectParts:
|
||||
parts, err := formUploadAttributes(info, p.MaxParts, p.PartNumberMarker)
|
||||
if err != nil {
|
||||
|
@ -219,10 +225,15 @@ func formUploadAttributes(info *data.ObjectInfo, maxParts, marker int) (*ObjectP
|
|||
if err != nil {
|
||||
return nil, fmt.Errorf("invalid completed part: %w", err)
|
||||
}
|
||||
// ETag value contains SHA256 checksum.
|
||||
checksumBytes, err := hex.DecodeString(part.ETag)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("invalid sha256 checksum in completed part: %w", err)
|
||||
}
|
||||
parts[i] = Part{
|
||||
PartNumber: part.PartNumber,
|
||||
Size: int(part.Size),
|
||||
ChecksumSHA256: part.ETag,
|
||||
ChecksumSHA256: base64.StdEncoding.EncodeToString(checksumBytes),
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
@ -1,6 +1,8 @@
|
|||
package handler
|
||||
|
||||
import (
|
||||
"encoding/base64"
|
||||
"encoding/hex"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
|
@ -24,11 +26,13 @@ func TestGetObjectPartsAttributes(t *testing.T) {
|
|||
multipartUpload := createMultipartUpload(hc, bktName, objMultipartName, map[string]string{})
|
||||
etag, _ := uploadPart(hc, bktName, objMultipartName, multipartUpload.UploadID, 1, partSize)
|
||||
completeMultipartUpload(hc, bktName, objMultipartName, multipartUpload.UploadID, []string{etag})
|
||||
etagBytes, err := hex.DecodeString(etag)
|
||||
require.NoError(t, err)
|
||||
|
||||
result = getObjectAttributes(hc, bktName, objMultipartName, objectParts)
|
||||
require.NotNil(t, result.ObjectParts)
|
||||
require.Len(t, result.ObjectParts.Parts, 1)
|
||||
require.Equal(t, etag, result.ObjectParts.Parts[0].ChecksumSHA256)
|
||||
require.Equal(t, base64.StdEncoding.EncodeToString(etagBytes), result.ObjectParts.Parts[0].ChecksumSHA256)
|
||||
require.Equal(t, partSize, result.ObjectParts.Parts[0].Size)
|
||||
require.Equal(t, 1, result.ObjectParts.PartsCount)
|
||||
}
|
||||
|
|
|
@ -164,7 +164,7 @@ func (h *handler) CopyObjectHandler(w http.ResponseWriter, r *http.Request) {
|
|||
}
|
||||
}
|
||||
|
||||
if err = checkPreconditions(srcObjInfo, args.Conditional); err != nil {
|
||||
if err = checkPreconditions(srcObjInfo, args.Conditional, h.cfg.Features.MD5Enabled()); err != nil {
|
||||
h.logAndSendError(w, "precondition failed", reqInfo, errors.GetAPIError(errors.ErrPreconditionFailed))
|
||||
return
|
||||
}
|
||||
|
@ -210,7 +210,7 @@ func (h *handler) CopyObjectHandler(w http.ResponseWriter, r *http.Request) {
|
|||
}
|
||||
dstObjInfo := extendedDstObjInfo.ObjectInfo
|
||||
|
||||
if err = middleware.EncodeToResponse(w, &CopyObjectResponse{LastModified: dstObjInfo.Created.UTC().Format(time.RFC3339), ETag: dstObjInfo.HashSum}); err != nil {
|
||||
if err = middleware.EncodeToResponse(w, &CopyObjectResponse{LastModified: dstObjInfo.Created.UTC().Format(time.RFC3339), ETag: dstObjInfo.ETag(h.cfg.Features.MD5Enabled())}); err != nil {
|
||||
h.logAndSendError(w, "something went wrong", reqInfo, err, additional...)
|
||||
return
|
||||
}
|
||||
|
@ -254,7 +254,7 @@ func (h *handler) CopyObjectHandler(w http.ResponseWriter, r *http.Request) {
|
|||
|
||||
s := &SendNotificationParams{
|
||||
Event: EventObjectCreatedCopy,
|
||||
NotificationInfo: data.NotificationInfoFromObject(dstObjInfo),
|
||||
NotificationInfo: data.NotificationInfoFromObject(dstObjInfo, h.cfg.Features.MD5Enabled()),
|
||||
BktInfo: dstBktInfo,
|
||||
ReqInfo: reqInfo,
|
||||
}
|
||||
|
|
|
@ -78,7 +78,8 @@ func addSSECHeaders(responseHeader http.Header, requestHeader http.Header) {
|
|||
responseHeader.Set(api.AmzServerSideEncryptionCustomerKeyMD5, requestHeader.Get(api.AmzServerSideEncryptionCustomerKeyMD5))
|
||||
}
|
||||
|
||||
func writeHeaders(h http.Header, requestHeader http.Header, extendedInfo *data.ExtendedObjectInfo, tagSetLength int, isBucketUnversioned bool) {
|
||||
func writeHeaders(h http.Header, requestHeader http.Header, extendedInfo *data.ExtendedObjectInfo, tagSetLength int,
|
||||
isBucketUnversioned, md5Enabled bool) {
|
||||
info := extendedInfo.ObjectInfo
|
||||
if len(info.ContentType) > 0 && h.Get(api.ContentType) == "" {
|
||||
h.Set(api.ContentType, info.ContentType)
|
||||
|
@ -94,7 +95,8 @@ func writeHeaders(h http.Header, requestHeader http.Header, extendedInfo *data.E
|
|||
h.Set(api.ContentLength, strconv.FormatUint(info.Size, 10))
|
||||
}
|
||||
|
||||
h.Set(api.ETag, info.HashSum)
|
||||
h.Set(api.ETag, info.ETag(md5Enabled))
|
||||
|
||||
h.Set(api.AmzTaggingCount, strconv.Itoa(tagSetLength))
|
||||
|
||||
if !isBucketUnversioned {
|
||||
|
@ -151,7 +153,7 @@ func (h *handler) GetObjectHandler(w http.ResponseWriter, r *http.Request) {
|
|||
}
|
||||
info := extendedInfo.ObjectInfo
|
||||
|
||||
if err = checkPreconditions(info, conditional); err != nil {
|
||||
if err = checkPreconditions(info, conditional, h.cfg.Features.MD5Enabled()); err != nil {
|
||||
h.logAndSendError(w, "precondition failed", reqInfo, err)
|
||||
return
|
||||
}
|
||||
|
@ -219,7 +221,7 @@ func (h *handler) GetObjectHandler(w http.ResponseWriter, r *http.Request) {
|
|||
return
|
||||
}
|
||||
|
||||
writeHeaders(w.Header(), r.Header, extendedInfo, len(tagSet), bktSettings.Unversioned())
|
||||
writeHeaders(w.Header(), r.Header, extendedInfo, len(tagSet), bktSettings.Unversioned(), h.cfg.Features.MD5Enabled())
|
||||
if params != nil {
|
||||
writeRangeHeaders(w, params, fullSize)
|
||||
} else {
|
||||
|
@ -232,12 +234,13 @@ func (h *handler) GetObjectHandler(w http.ResponseWriter, r *http.Request) {
|
|||
}
|
||||
}
|
||||
|
||||
func checkPreconditions(info *data.ObjectInfo, args *conditionalArgs) error {
|
||||
if len(args.IfMatch) > 0 && args.IfMatch != info.HashSum {
|
||||
return fmt.Errorf("%w: etag mismatched: '%s', '%s'", errors.GetAPIError(errors.ErrPreconditionFailed), args.IfMatch, info.HashSum)
|
||||
func checkPreconditions(info *data.ObjectInfo, args *conditionalArgs, md5Enabled bool) error {
|
||||
etag := info.ETag(md5Enabled)
|
||||
if len(args.IfMatch) > 0 && args.IfMatch != etag {
|
||||
return fmt.Errorf("%w: etag mismatched: '%s', '%s'", errors.GetAPIError(errors.ErrPreconditionFailed), args.IfMatch, etag)
|
||||
}
|
||||
if len(args.IfNoneMatch) > 0 && args.IfNoneMatch == info.HashSum {
|
||||
return fmt.Errorf("%w: etag matched: '%s', '%s'", errors.GetAPIError(errors.ErrNotModified), args.IfNoneMatch, info.HashSum)
|
||||
if len(args.IfNoneMatch) > 0 && args.IfNoneMatch == etag {
|
||||
return fmt.Errorf("%w: etag matched: '%s', '%s'", errors.GetAPIError(errors.ErrNotModified), args.IfNoneMatch, etag)
|
||||
}
|
||||
if args.IfModifiedSince != nil && info.Created.Before(*args.IfModifiedSince) {
|
||||
return fmt.Errorf("%w: not modified since '%s', last modified '%s'", errors.GetAPIError(errors.ErrNotModified),
|
||||
|
|
|
@ -147,7 +147,7 @@ func TestPreconditions(t *testing.T) {
|
|||
},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
actual := checkPreconditions(tc.info, tc.args)
|
||||
actual := checkPreconditions(tc.info, tc.args, false)
|
||||
if tc.expected == nil {
|
||||
require.NoError(t, actual)
|
||||
} else {
|
||||
|
@ -197,6 +197,19 @@ func TestGetObject(t *testing.T) {
|
|||
getObjectAssertS3Error(hc, bktName, objName, emptyVersion, errors.ErrNoSuchKey)
|
||||
}
|
||||
|
||||
func TestGetObjectEnabledMD5(t *testing.T) {
|
||||
hc := prepareHandlerContext(t)
|
||||
bktName, objName := "bucket", "obj"
|
||||
_, objInfo := createBucketAndObject(hc, bktName, objName)
|
||||
|
||||
_, headers := getObject(hc, bktName, objName)
|
||||
require.Equal(t, objInfo.HashSum, headers.Get(api.ETag))
|
||||
|
||||
hc.features.SetMD5Enabled(true)
|
||||
_, headers = getObject(hc, bktName, objName)
|
||||
require.Equal(t, objInfo.MD5Sum, headers.Get(api.ETag))
|
||||
}
|
||||
|
||||
func putObjectContent(hc *handlerContext, bktName, objName, content string) {
|
||||
body := bytes.NewReader([]byte(content))
|
||||
w, r := prepareTestPayloadRequest(hc, bktName, objName, body)
|
||||
|
|
|
@ -39,7 +39,7 @@ type handlerContext struct {
|
|||
context context.Context
|
||||
kludge *kludgeSettingsMock
|
||||
|
||||
layerFeatures *layer.FeatureSettingsMock
|
||||
features *layer.FeatureSettingsMock
|
||||
}
|
||||
|
||||
func (hc *handlerContext) Handler() *handler {
|
||||
|
@ -148,6 +148,7 @@ func prepareHandlerContextBase(t *testing.T, minCache bool) *handlerContext {
|
|||
Policy: &placementPolicyMock{defaultPolicy: pp},
|
||||
XMLDecoder: &xmlDecoderProviderMock{},
|
||||
Kludge: kludge,
|
||||
Features: features,
|
||||
},
|
||||
}
|
||||
|
||||
|
@ -160,7 +161,7 @@ func prepareHandlerContextBase(t *testing.T, minCache bool) *handlerContext {
|
|||
context: middleware.SetBoxData(context.Background(), newTestAccessBox(t, key)),
|
||||
kludge: kludge,
|
||||
|
||||
layerFeatures: features,
|
||||
features: features,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
@ -65,7 +65,7 @@ func (h *handler) HeadObjectHandler(w http.ResponseWriter, r *http.Request) {
|
|||
return
|
||||
}
|
||||
|
||||
if err = checkPreconditions(info, conditional); err != nil {
|
||||
if err = checkPreconditions(info, conditional, h.cfg.Features.MD5Enabled()); err != nil {
|
||||
h.logAndSendError(w, "precondition failed", reqInfo, err)
|
||||
return
|
||||
}
|
||||
|
@ -118,7 +118,7 @@ func (h *handler) HeadObjectHandler(w http.ResponseWriter, r *http.Request) {
|
|||
return
|
||||
}
|
||||
|
||||
writeHeaders(w.Header(), r.Header, extendedInfo, len(tagSet), bktSettings.Unversioned())
|
||||
writeHeaders(w.Header(), r.Header, extendedInfo, len(tagSet), bktSettings.Unversioned(), h.cfg.Features.MD5Enabled())
|
||||
w.WriteHeader(http.StatusOK)
|
||||
}
|
||||
|
||||
|
|
|
@ -243,6 +243,7 @@ func (h *handler) UploadPartHandler(w http.ResponseWriter, r *http.Request) {
|
|||
PartNumber: partNumber,
|
||||
Size: size,
|
||||
Reader: body,
|
||||
ContentMD5: r.Header.Get(api.ContentMD5),
|
||||
}
|
||||
|
||||
p.Info.Encryption, err = formEncryptionParams(r)
|
||||
|
@ -336,7 +337,7 @@ func (h *handler) UploadPartCopy(w http.ResponseWriter, r *http.Request) {
|
|||
return
|
||||
}
|
||||
|
||||
if err = checkPreconditions(srcInfo, args.Conditional); err != nil {
|
||||
if err = checkPreconditions(srcInfo, args.Conditional, h.cfg.Features.MD5Enabled()); err != nil {
|
||||
h.logAndSendError(w, "precondition failed", reqInfo, errors.GetAPIError(errors.ErrPreconditionFailed),
|
||||
additional...)
|
||||
return
|
||||
|
@ -373,8 +374,8 @@ func (h *handler) UploadPartCopy(w http.ResponseWriter, r *http.Request) {
|
|||
}
|
||||
|
||||
response := UploadPartCopyResponse{
|
||||
ETag: info.HashSum,
|
||||
LastModified: info.Created.UTC().Format(time.RFC3339),
|
||||
ETag: info.ETag(h.cfg.Features.MD5Enabled()),
|
||||
}
|
||||
|
||||
if p.Info.Encryption.Enabled() {
|
||||
|
@ -449,8 +450,8 @@ func (h *handler) CompleteMultipartUploadHandler(w http.ResponseWriter, r *http.
|
|||
|
||||
response := CompleteMultipartUploadResponse{
|
||||
Bucket: objInfo.Bucket,
|
||||
ETag: objInfo.HashSum,
|
||||
Key: objInfo.Name,
|
||||
ETag: objInfo.ETag(h.cfg.Features.MD5Enabled()),
|
||||
}
|
||||
|
||||
// Here we previously set api.AmzVersionID header for versioned bucket.
|
||||
|
@ -514,7 +515,7 @@ func (h *handler) completeMultipartUpload(r *http.Request, c *layer.CompleteMult
|
|||
|
||||
s := &SendNotificationParams{
|
||||
Event: EventObjectCreatedCompleteMultipartUpload,
|
||||
NotificationInfo: data.NotificationInfoFromObject(objInfo),
|
||||
NotificationInfo: data.NotificationInfoFromObject(objInfo, h.cfg.Features.MD5Enabled()),
|
||||
BktInfo: bktInfo,
|
||||
ReqInfo: reqInfo,
|
||||
}
|
||||
|
|
|
@ -2,6 +2,8 @@ package handler
|
|||
|
||||
import (
|
||||
"bytes"
|
||||
"crypto/md5"
|
||||
"encoding/hex"
|
||||
"encoding/xml"
|
||||
"fmt"
|
||||
"net/http"
|
||||
|
@ -255,6 +257,32 @@ func TestListParts(t *testing.T) {
|
|||
require.Len(t, list.Parts, 0)
|
||||
}
|
||||
|
||||
func TestMultipartUploadEnabledMD5(t *testing.T) {
|
||||
hc := prepareHandlerContext(t)
|
||||
hc.features.SetMD5Enabled(true)
|
||||
|
||||
bktName, objName := "bucket-md5", "object-md5"
|
||||
createTestBucket(hc, bktName)
|
||||
|
||||
partSize := 5 * 1024 * 1024
|
||||
multipartUpload := createMultipartUpload(hc, bktName, objName, map[string]string{})
|
||||
etag1, partBody1 := uploadPart(hc, bktName, objName, multipartUpload.UploadID, 1, partSize)
|
||||
md5Sum1 := md5.Sum(partBody1)
|
||||
require.Equal(t, hex.EncodeToString(md5Sum1[:]), etag1)
|
||||
|
||||
etag2, partBody2 := uploadPart(hc, bktName, objName, multipartUpload.UploadID, 2, partSize)
|
||||
md5Sum2 := md5.Sum(partBody2)
|
||||
require.Equal(t, hex.EncodeToString(md5Sum2[:]), etag2)
|
||||
|
||||
w := completeMultipartUploadBase(hc, bktName, objName, multipartUpload.UploadID, []string{etag1, etag2})
|
||||
assertStatus(t, w, http.StatusOK)
|
||||
resp := &CompleteMultipartUploadResponse{}
|
||||
err := xml.NewDecoder(w.Result().Body).Decode(resp)
|
||||
require.NoError(t, err)
|
||||
completeMD5Sum := md5.Sum(append(md5Sum1[:], md5Sum2[:]...))
|
||||
require.Equal(t, hex.EncodeToString(completeMD5Sum[:])+"-2", resp.ETag)
|
||||
}
|
||||
|
||||
func uploadPartCopy(hc *handlerContext, bktName, objName, uploadID string, num int, srcObj string, start, end int) *UploadPartCopyResponse {
|
||||
return uploadPartCopyBase(hc, bktName, objName, false, uploadID, num, srcObj, start, end)
|
||||
}
|
||||
|
|
|
@ -33,12 +33,12 @@ func (h *handler) ListObjectsV1Handler(w http.ResponseWriter, r *http.Request) {
|
|||
return
|
||||
}
|
||||
|
||||
if err = middleware.EncodeToResponse(w, encodeV1(params, list)); err != nil {
|
||||
if err = middleware.EncodeToResponse(w, h.encodeV1(params, list)); err != nil {
|
||||
h.logAndSendError(w, "something went wrong", reqInfo, err)
|
||||
}
|
||||
}
|
||||
|
||||
func encodeV1(p *layer.ListObjectsParamsV1, list *layer.ListObjectsInfoV1) *ListObjectsV1Response {
|
||||
func (h *handler) encodeV1(p *layer.ListObjectsParamsV1, list *layer.ListObjectsInfoV1) *ListObjectsV1Response {
|
||||
res := &ListObjectsV1Response{
|
||||
Name: p.BktInfo.Name,
|
||||
EncodingType: p.Encode,
|
||||
|
@ -52,7 +52,7 @@ func encodeV1(p *layer.ListObjectsParamsV1, list *layer.ListObjectsInfoV1) *List
|
|||
|
||||
res.CommonPrefixes = fillPrefixes(list.Prefixes, p.Encode)
|
||||
|
||||
res.Contents = fillContentsWithOwner(list.Objects, p.Encode)
|
||||
res.Contents = fillContentsWithOwner(list.Objects, p.Encode, h.cfg.Features.MD5Enabled())
|
||||
|
||||
return res
|
||||
}
|
||||
|
@ -77,12 +77,12 @@ func (h *handler) ListObjectsV2Handler(w http.ResponseWriter, r *http.Request) {
|
|||
return
|
||||
}
|
||||
|
||||
if err = middleware.EncodeToResponse(w, encodeV2(params, list)); err != nil {
|
||||
if err = middleware.EncodeToResponse(w, h.encodeV2(params, list)); err != nil {
|
||||
h.logAndSendError(w, "something went wrong", reqInfo, err)
|
||||
}
|
||||
}
|
||||
|
||||
func encodeV2(p *layer.ListObjectsParamsV2, list *layer.ListObjectsInfoV2) *ListObjectsV2Response {
|
||||
func (h *handler) encodeV2(p *layer.ListObjectsParamsV2, list *layer.ListObjectsInfoV2) *ListObjectsV2Response {
|
||||
res := &ListObjectsV2Response{
|
||||
Name: p.BktInfo.Name,
|
||||
EncodingType: p.Encode,
|
||||
|
@ -98,7 +98,7 @@ func encodeV2(p *layer.ListObjectsParamsV2, list *layer.ListObjectsInfoV2) *List
|
|||
|
||||
res.CommonPrefixes = fillPrefixes(list.Prefixes, p.Encode)
|
||||
|
||||
res.Contents = fillContents(list.Objects, p.Encode, p.FetchOwner)
|
||||
res.Contents = fillContents(list.Objects, p.Encode, p.FetchOwner, h.cfg.Features.MD5Enabled())
|
||||
|
||||
return res
|
||||
}
|
||||
|
@ -184,18 +184,18 @@ func fillPrefixes(src []string, encode string) []CommonPrefix {
|
|||
return dst
|
||||
}
|
||||
|
||||
func fillContentsWithOwner(src []*data.ObjectInfo, encode string) []Object {
|
||||
return fillContents(src, encode, true)
|
||||
func fillContentsWithOwner(src []*data.ObjectInfo, encode string, md5Enabled bool) []Object {
|
||||
return fillContents(src, encode, true, md5Enabled)
|
||||
}
|
||||
|
||||
func fillContents(src []*data.ObjectInfo, encode string, fetchOwner bool) []Object {
|
||||
func fillContents(src []*data.ObjectInfo, encode string, fetchOwner, md5Enabled bool) []Object {
|
||||
var dst []Object
|
||||
for _, obj := range src {
|
||||
res := Object{
|
||||
Key: s3PathEncode(obj.Name, encode),
|
||||
Size: obj.Size,
|
||||
LastModified: obj.Created.UTC().Format(time.RFC3339),
|
||||
ETag: obj.HashSum,
|
||||
ETag: obj.ETag(md5Enabled),
|
||||
}
|
||||
|
||||
if size, err := layer.GetObjectSize(obj); err == nil {
|
||||
|
@ -233,7 +233,7 @@ func (h *handler) ListBucketObjectVersionsHandler(w http.ResponseWriter, r *http
|
|||
return
|
||||
}
|
||||
|
||||
response := encodeListObjectVersionsToResponse(info, p.BktInfo.Name)
|
||||
response := encodeListObjectVersionsToResponse(info, p.BktInfo.Name, h.cfg.Features.MD5Enabled())
|
||||
if err = middleware.EncodeToResponse(w, response); err != nil {
|
||||
h.logAndSendError(w, "something went wrong", reqInfo, err)
|
||||
}
|
||||
|
@ -261,7 +261,7 @@ func parseListObjectVersionsRequest(reqInfo *middleware.ReqInfo) (*layer.ListObj
|
|||
return &res, nil
|
||||
}
|
||||
|
||||
func encodeListObjectVersionsToResponse(info *layer.ListObjectVersionsInfo, bucketName string) *ListObjectsVersionsResponse {
|
||||
func encodeListObjectVersionsToResponse(info *layer.ListObjectVersionsInfo, bucketName string, md5Enabled bool) *ListObjectsVersionsResponse {
|
||||
res := ListObjectsVersionsResponse{
|
||||
Name: bucketName,
|
||||
IsTruncated: info.IsTruncated,
|
||||
|
@ -286,7 +286,7 @@ func encodeListObjectVersionsToResponse(info *layer.ListObjectVersionsInfo, buck
|
|||
},
|
||||
Size: ver.ObjectInfo.Size,
|
||||
VersionID: ver.Version(),
|
||||
ETag: ver.ObjectInfo.HashSum,
|
||||
ETag: ver.ObjectInfo.ETag(md5Enabled),
|
||||
})
|
||||
}
|
||||
// this loop is not starting till versioning is not implemented
|
||||
|
|
|
@ -242,6 +242,7 @@ func (h *handler) PutObjectHandler(w http.ResponseWriter, r *http.Request) {
|
|||
Size: size,
|
||||
Header: metadata,
|
||||
Encryption: encryptionParams,
|
||||
ContentMD5: r.Header.Get(api.ContentMD5),
|
||||
}
|
||||
|
||||
params.CopiesNumbers, err = h.pickCopiesNumbers(metadata, bktInfo.LocationConstraint)
|
||||
|
@ -273,7 +274,7 @@ func (h *handler) PutObjectHandler(w http.ResponseWriter, r *http.Request) {
|
|||
|
||||
s := &SendNotificationParams{
|
||||
Event: EventObjectCreatedPut,
|
||||
NotificationInfo: data.NotificationInfoFromObject(objInfo),
|
||||
NotificationInfo: data.NotificationInfoFromObject(objInfo, h.cfg.Features.MD5Enabled()),
|
||||
BktInfo: bktInfo,
|
||||
ReqInfo: reqInfo,
|
||||
}
|
||||
|
@ -324,7 +325,8 @@ func (h *handler) PutObjectHandler(w http.ResponseWriter, r *http.Request) {
|
|||
addSSECHeaders(w.Header(), r.Header)
|
||||
}
|
||||
|
||||
w.Header().Set(api.ETag, objInfo.HashSum)
|
||||
w.Header().Set(api.ETag, objInfo.ETag(h.cfg.Features.MD5Enabled()))
|
||||
|
||||
middleware.WriteSuccessResponseHeadersOnly(w)
|
||||
}
|
||||
|
||||
|
@ -490,7 +492,7 @@ func (h *handler) PostObject(w http.ResponseWriter, r *http.Request) {
|
|||
|
||||
s := &SendNotificationParams{
|
||||
Event: EventObjectCreatedPost,
|
||||
NotificationInfo: data.NotificationInfoFromObject(objInfo),
|
||||
NotificationInfo: data.NotificationInfoFromObject(objInfo, h.cfg.Features.MD5Enabled()),
|
||||
BktInfo: bktInfo,
|
||||
ReqInfo: reqInfo,
|
||||
}
|
||||
|
@ -559,7 +561,7 @@ func (h *handler) PostObject(w http.ResponseWriter, r *http.Request) {
|
|||
resp := &PostResponse{
|
||||
Bucket: objInfo.Bucket,
|
||||
Key: objInfo.Name,
|
||||
ETag: objInfo.HashSum,
|
||||
ETag: objInfo.ETag(h.cfg.Features.MD5Enabled()),
|
||||
}
|
||||
w.WriteHeader(status)
|
||||
if _, err = w.Write(middleware.EncodeResponse(resp)); err != nil {
|
||||
|
@ -569,7 +571,7 @@ func (h *handler) PostObject(w http.ResponseWriter, r *http.Request) {
|
|||
}
|
||||
}
|
||||
|
||||
w.Header().Set(api.ETag, objInfo.HashSum)
|
||||
w.Header().Set(api.ETag, objInfo.ETag(h.cfg.Features.MD5Enabled()))
|
||||
w.WriteHeader(status)
|
||||
}
|
||||
|
||||
|
|
|
@ -3,7 +3,10 @@ package handler
|
|||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/md5"
|
||||
"crypto/rand"
|
||||
"encoding/base64"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"io"
|
||||
|
@ -194,6 +197,37 @@ func TestPutObjectWithWrapReaderDiscardOnError(t *testing.T) {
|
|||
require.Equal(t, numGoroutineBefore, numGoroutineAfter, "goroutines shouldn't leak during put object")
|
||||
}
|
||||
|
||||
func TestPutObjectWithInvalidContentMD5(t *testing.T) {
|
||||
tc := prepareHandlerContext(t)
|
||||
tc.features.SetMD5Enabled(true)
|
||||
|
||||
bktName, objName := "bucket-for-put", "object-for-put"
|
||||
createTestBucket(tc, bktName)
|
||||
|
||||
content := []byte("content")
|
||||
w, r := prepareTestPayloadRequest(tc, bktName, objName, bytes.NewReader(content))
|
||||
r.Header.Set(api.ContentMD5, base64.StdEncoding.EncodeToString([]byte("invalid")))
|
||||
tc.Handler().PutObjectHandler(w, r)
|
||||
assertS3Error(t, w, s3errors.GetAPIError(s3errors.ErrInvalidDigest))
|
||||
|
||||
checkNotFound(t, tc, bktName, objName, emptyVersion)
|
||||
}
|
||||
|
||||
func TestPutObjectWithEnabledMD5(t *testing.T) {
|
||||
tc := prepareHandlerContext(t)
|
||||
tc.features.SetMD5Enabled(true)
|
||||
|
||||
bktName, objName := "bucket-for-put", "object-for-put"
|
||||
createTestBucket(tc, bktName)
|
||||
|
||||
content := []byte("content")
|
||||
md5Hash := md5.New()
|
||||
md5Hash.Write(content)
|
||||
w, r := prepareTestPayloadRequest(tc, bktName, objName, bytes.NewReader(content))
|
||||
tc.Handler().PutObjectHandler(w, r)
|
||||
require.Equal(t, hex.EncodeToString(md5Hash.Sum(nil)), w.Header().Get(api.ETag))
|
||||
}
|
||||
|
||||
func TestPutObjectWithStreamBodyAWSExample(t *testing.T) {
|
||||
hc := prepareHandlerContext(t)
|
||||
|
||||
|
@ -320,7 +354,7 @@ func TestPutObjectClientCut(t *testing.T) {
|
|||
obj1 := getObjectFromLayer(hc, objName1)[0]
|
||||
require.Empty(t, getObjectAttribute(obj1, "s3-client-cut"))
|
||||
|
||||
hc.layerFeatures.SetClientCut(true)
|
||||
hc.features.SetClientCut(true)
|
||||
putObject(hc, bktName, objName2)
|
||||
obj2 := getObjectFromLayer(hc, objName2)[0]
|
||||
require.Equal(t, "true", getObjectAttribute(obj2, "s3-client-cut"))
|
||||
|
|
|
@ -45,7 +45,7 @@ func (n *layer) PutBucketCORS(ctx context.Context, p *PutCORSParams) error {
|
|||
CopiesNumber: p.CopiesNumbers,
|
||||
}
|
||||
|
||||
_, objID, _, err := n.objectPutAndHash(ctx, prm, p.BktInfo)
|
||||
_, objID, _, _, err := n.objectPutAndHash(ctx, prm, p.BktInfo)
|
||||
if err != nil {
|
||||
return fmt.Errorf("put system object: %w", err)
|
||||
}
|
||||
|
|
|
@ -27,6 +27,7 @@ import (
|
|||
|
||||
type FeatureSettingsMock struct {
|
||||
clientCut bool
|
||||
md5Enabled bool
|
||||
}
|
||||
|
||||
func (k *FeatureSettingsMock) ClientCut() bool {
|
||||
|
@ -37,6 +38,14 @@ func (k *FeatureSettingsMock) SetClientCut(clientCut bool) {
|
|||
k.clientCut = clientCut
|
||||
}
|
||||
|
||||
func (k *FeatureSettingsMock) MD5Enabled() bool {
|
||||
return k.md5Enabled
|
||||
}
|
||||
|
||||
func (k *FeatureSettingsMock) SetMD5Enabled(md5Enabled bool) {
|
||||
k.md5Enabled = md5Enabled
|
||||
}
|
||||
|
||||
type TestFrostFS struct {
|
||||
FrostFS
|
||||
|
||||
|
|
|
@ -48,6 +48,7 @@ type (
|
|||
|
||||
FeatureSettings interface {
|
||||
ClientCut() bool
|
||||
MD5Enabled() bool
|
||||
}
|
||||
|
||||
layer struct {
|
||||
|
@ -117,6 +118,8 @@ type (
|
|||
Lock *data.ObjectLock
|
||||
Encryption encryption.Params
|
||||
CopiesNumbers []uint32
|
||||
CompleteMD5Hash string
|
||||
ContentMD5 string
|
||||
}
|
||||
|
||||
PutCombinedObjectParams struct {
|
||||
|
|
|
@ -3,6 +3,8 @@ package layer
|
|||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/md5"
|
||||
"encoding/base64"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
|
@ -68,6 +70,7 @@ type (
|
|||
PartNumber int
|
||||
Size uint64
|
||||
Reader io.Reader
|
||||
ContentMD5 string
|
||||
}
|
||||
|
||||
UploadCopyParams struct {
|
||||
|
@ -197,7 +200,7 @@ func (n *layer) UploadPart(ctx context.Context, p *UploadPartParams) (string, er
|
|||
return "", err
|
||||
}
|
||||
|
||||
return objInfo.HashSum, nil
|
||||
return objInfo.ETag(n.features.MD5Enabled()), nil
|
||||
}
|
||||
|
||||
func (n *layer) uploadPart(ctx context.Context, multipartInfo *data.MultipartInfo, p *UploadPartParams) (*data.ObjectInfo, error) {
|
||||
|
@ -230,10 +233,28 @@ func (n *layer) uploadPart(ctx context.Context, multipartInfo *data.MultipartInf
|
|||
prm.Attributes[0][0], prm.Attributes[0][1] = UploadIDAttributeName, p.Info.UploadID
|
||||
prm.Attributes[1][0], prm.Attributes[1][1] = UploadPartNumberAttributeName, strconv.Itoa(p.PartNumber)
|
||||
|
||||
size, id, hash, err := n.objectPutAndHash(ctx, prm, bktInfo)
|
||||
size, id, hash, md5Hash, err := n.objectPutAndHash(ctx, prm, bktInfo)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if len(p.ContentMD5) > 0 {
|
||||
hashBytes, err := base64.StdEncoding.DecodeString(p.ContentMD5)
|
||||
if err != nil {
|
||||
return nil, s3errors.GetAPIError(s3errors.ErrInvalidDigest)
|
||||
}
|
||||
if hex.EncodeToString(hashBytes) != hex.EncodeToString(md5Hash) {
|
||||
prm := PrmObjectDelete{
|
||||
Object: id,
|
||||
Container: bktInfo.CID,
|
||||
}
|
||||
n.prepareAuthParameters(ctx, &prm.PrmAuth, bktInfo.Owner)
|
||||
err = n.frostFS.DeleteObject(ctx, prm)
|
||||
if err != nil {
|
||||
n.reqLogger(ctx).Debug(logs.FailedToDeleteObject, zap.Stringer("cid", bktInfo.CID), zap.Stringer("oid", id))
|
||||
}
|
||||
return nil, s3errors.GetAPIError(s3errors.ErrInvalidDigest)
|
||||
}
|
||||
}
|
||||
if p.Info.Encryption.Enabled() {
|
||||
size = decSize
|
||||
}
|
||||
|
@ -250,6 +271,7 @@ func (n *layer) uploadPart(ctx context.Context, multipartInfo *data.MultipartInf
|
|||
Size: size,
|
||||
ETag: hex.EncodeToString(hash),
|
||||
Created: prm.CreationTime,
|
||||
MD5: hex.EncodeToString(md5Hash),
|
||||
}
|
||||
|
||||
oldPartID, err := n.treeService.AddPart(ctx, bktInfo, multipartInfo.ID, partInfo)
|
||||
|
@ -274,6 +296,7 @@ func (n *layer) uploadPart(ctx context.Context, multipartInfo *data.MultipartInf
|
|||
Size: partInfo.Size,
|
||||
Created: partInfo.Created,
|
||||
HashSum: partInfo.ETag,
|
||||
MD5Sum: partInfo.MD5,
|
||||
}
|
||||
|
||||
return objInfo, nil
|
||||
|
@ -347,9 +370,10 @@ func (n *layer) CompleteMultipartUpload(ctx context.Context, p *CompleteMultipar
|
|||
parts := make([]*data.PartInfo, 0, len(p.Parts))
|
||||
|
||||
var completedPartsHeader strings.Builder
|
||||
md5Hash := md5.New()
|
||||
for i, part := range p.Parts {
|
||||
partInfo := partsInfo[part.PartNumber]
|
||||
if partInfo == nil || part.ETag != partInfo.ETag {
|
||||
if partInfo == nil || strings.Trim(part.ETag, "\"") != partInfo.GetETag(n.features.MD5Enabled()) {
|
||||
return nil, nil, fmt.Errorf("%w: unknown part %d or etag mismatched", s3errors.GetAPIError(s3errors.ErrInvalidPart), part.PartNumber)
|
||||
}
|
||||
delete(partsInfo, part.PartNumber)
|
||||
|
@ -376,6 +400,12 @@ func (n *layer) CompleteMultipartUpload(ctx context.Context, p *CompleteMultipar
|
|||
if _, err = completedPartsHeader.WriteString(partInfoStr); err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
bytesHash, err := hex.DecodeString(partInfo.MD5)
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("couldn't decode MD5 checksum of part: %w", err)
|
||||
}
|
||||
md5Hash.Write(bytesHash)
|
||||
}
|
||||
|
||||
initMetadata := make(map[string]string, len(multipartInfo.Meta)+1)
|
||||
|
@ -417,6 +447,7 @@ func (n *layer) CompleteMultipartUpload(ctx context.Context, p *CompleteMultipar
|
|||
Size: multipartObjetSize,
|
||||
Encryption: p.Info.Encryption,
|
||||
CopiesNumbers: multipartInfo.CopiesNumbers,
|
||||
CompleteMD5Hash: hex.EncodeToString(md5Hash.Sum(nil)) + "-" + strconv.Itoa(len(p.Parts)),
|
||||
})
|
||||
if err != nil {
|
||||
n.reqLogger(ctx).Error(logs.CouldNotPutCompletedObject,
|
||||
|
@ -537,7 +568,7 @@ func (n *layer) ListParts(ctx context.Context, p *ListPartsParams) (*ListPartsIn
|
|||
|
||||
for _, partInfo := range partsInfo {
|
||||
parts = append(parts, &Part{
|
||||
ETag: partInfo.ETag,
|
||||
ETag: partInfo.GetETag(n.features.MD5Enabled()),
|
||||
LastModified: partInfo.Created.UTC().Format(time.RFC3339),
|
||||
PartNumber: partInfo.Number,
|
||||
Size: partInfo.Size,
|
||||
|
|
|
@ -34,7 +34,7 @@ func (n *layer) PutBucketNotificationConfiguration(ctx context.Context, p *PutBu
|
|||
CopiesNumber: p.CopiesNumbers,
|
||||
}
|
||||
|
||||
_, objID, _, err := n.objectPutAndHash(ctx, prm, p.BktInfo)
|
||||
_, objID, _, _, err := n.objectPutAndHash(ctx, prm, p.BktInfo)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
|
|
@ -1,8 +1,11 @@
|
|||
package layer
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/md5"
|
||||
"crypto/sha256"
|
||||
"encoding/base64"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
|
@ -287,10 +290,23 @@ func (n *layer) PutObject(ctx context.Context, p *PutObjectParams) (*data.Extend
|
|||
prm.Attributes = append(prm.Attributes, [2]string{k, v})
|
||||
}
|
||||
|
||||
size, id, hash, err := n.objectPutAndHash(ctx, prm, p.BktInfo)
|
||||
size, id, hash, md5Hash, err := n.objectPutAndHash(ctx, prm, p.BktInfo)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if len(p.ContentMD5) > 0 {
|
||||
headerMd5Hash, err := base64.StdEncoding.DecodeString(p.ContentMD5)
|
||||
if err != nil {
|
||||
return nil, apiErrors.GetAPIError(apiErrors.ErrInvalidDigest)
|
||||
}
|
||||
if !bytes.Equal(headerMd5Hash, md5Hash) {
|
||||
err = n.objectDelete(ctx, p.BktInfo, id)
|
||||
if err != nil {
|
||||
n.reqLogger(ctx).Debug(logs.FailedToDeleteObject, zap.Stringer("cid", p.BktInfo.CID), zap.Stringer("oid", id))
|
||||
}
|
||||
return nil, apiErrors.GetAPIError(apiErrors.ErrInvalidDigest)
|
||||
}
|
||||
}
|
||||
|
||||
n.reqLogger(ctx).Debug(logs.PutObject, zap.Stringer("cid", p.BktInfo.CID), zap.Stringer("oid", id))
|
||||
|
||||
|
@ -304,6 +320,11 @@ func (n *layer) PutObject(ctx context.Context, p *PutObjectParams) (*data.Extend
|
|||
IsUnversioned: !bktSettings.VersioningEnabled(),
|
||||
IsCombined: p.Header[MultipartObjectSize] != "",
|
||||
}
|
||||
if len(p.CompleteMD5Hash) > 0 {
|
||||
newVersion.MD5 = p.CompleteMD5Hash
|
||||
} else {
|
||||
newVersion.MD5 = hex.EncodeToString(md5Hash)
|
||||
}
|
||||
|
||||
if newVersion.ID, err = n.treeService.AddVersion(ctx, p.BktInfo, newVersion); err != nil {
|
||||
return nil, fmt.Errorf("couldn't add new verion to tree service: %w", err)
|
||||
|
@ -340,6 +361,7 @@ func (n *layer) PutObject(ctx context.Context, p *PutObjectParams) (*data.Extend
|
|||
Headers: p.Header,
|
||||
ContentType: p.Header[api.ContentType],
|
||||
HashSum: newVersion.ETag,
|
||||
MD5Sum: newVersion.MD5,
|
||||
}
|
||||
|
||||
extendedObjInfo := &data.ExtendedObjectInfo{
|
||||
|
@ -378,6 +400,7 @@ func (n *layer) headLastVersionIfNotDeleted(ctx context.Context, bkt *data.Bucke
|
|||
return nil, err
|
||||
}
|
||||
objInfo := objectInfoFromMeta(bkt, meta)
|
||||
objInfo.MD5Sum = node.MD5
|
||||
|
||||
extObjInfo := &data.ExtendedObjectInfo{
|
||||
ObjectInfo: objInfo,
|
||||
|
@ -430,6 +453,7 @@ func (n *layer) headVersion(ctx context.Context, bkt *data.BucketInfo, p *HeadOb
|
|||
return nil, err
|
||||
}
|
||||
objInfo := objectInfoFromMeta(bkt, meta)
|
||||
objInfo.MD5Sum = foundVersion.MD5
|
||||
|
||||
extObjInfo := &data.ExtendedObjectInfo{
|
||||
ObjectInfo: objInfo,
|
||||
|
@ -457,14 +481,16 @@ func (n *layer) objectDelete(ctx context.Context, bktInfo *data.BucketInfo, idOb
|
|||
|
||||
// objectPutAndHash prepare auth parameters and invoke frostfs.CreateObject.
|
||||
// Returns object ID and payload sha256 hash.
|
||||
func (n *layer) objectPutAndHash(ctx context.Context, prm PrmObjectCreate, bktInfo *data.BucketInfo) (uint64, oid.ID, []byte, error) {
|
||||
func (n *layer) objectPutAndHash(ctx context.Context, prm PrmObjectCreate, bktInfo *data.BucketInfo) (uint64, oid.ID, []byte, []byte, error) {
|
||||
n.prepareAuthParameters(ctx, &prm.PrmAuth, bktInfo.Owner)
|
||||
prm.ClientCut = n.features.ClientCut()
|
||||
var size uint64
|
||||
hash := sha256.New()
|
||||
md5Hash := md5.New()
|
||||
prm.Payload = wrapReader(prm.Payload, 64*1024, func(buf []byte) {
|
||||
size += uint64(len(buf))
|
||||
hash.Write(buf)
|
||||
md5Hash.Write(buf)
|
||||
})
|
||||
id, err := n.frostFS.CreateObject(ctx, prm)
|
||||
if err != nil {
|
||||
|
@ -472,9 +498,9 @@ func (n *layer) objectPutAndHash(ctx context.Context, prm PrmObjectCreate, bktIn
|
|||
n.reqLogger(ctx).Warn(logs.FailedToDiscardPutPayloadProbablyGoroutineLeaks, zap.Error(errDiscard))
|
||||
}
|
||||
|
||||
return 0, oid.ID{}, nil, err
|
||||
return 0, oid.ID{}, nil, nil, err
|
||||
}
|
||||
return size, id, hash.Sum(nil), nil
|
||||
return size, id, hash.Sum(nil), md5Hash.Sum(nil), nil
|
||||
}
|
||||
|
||||
// ListObjectsV1 returns objects in a bucket for requests of Version 1.
|
||||
|
@ -805,6 +831,7 @@ func (n *layer) objectInfoFromObjectsCacheOrFrostFS(ctx context.Context, bktInfo
|
|||
}
|
||||
|
||||
oi = objectInfoFromMeta(bktInfo, meta)
|
||||
oi.MD5Sum = node.MD5
|
||||
n.cache.PutObject(owner, &data.ExtendedObjectInfo{ObjectInfo: oi, NodeVersion: node})
|
||||
|
||||
return oi
|
||||
|
|
|
@ -125,7 +125,7 @@ func (n *layer) putLockObject(ctx context.Context, bktInfo *data.BucketInfo, obj
|
|||
return oid.ID{}, err
|
||||
}
|
||||
|
||||
_, id, _, err := n.objectPutAndHash(ctx, prm, bktInfo)
|
||||
_, id, _, _, err := n.objectPutAndHash(ctx, prm, bktInfo)
|
||||
return id, err
|
||||
}
|
||||
|
||||
|
|
|
@ -73,6 +73,7 @@ type (
|
|||
maxClient maxClientsConfig
|
||||
bypassContentEncodingInChunks atomic.Bool
|
||||
clientCut atomic.Bool
|
||||
md5Enabled atomic.Bool
|
||||
}
|
||||
|
||||
maxClientsConfig struct {
|
||||
|
@ -176,6 +177,7 @@ func newAppSettings(log *Logger, v *viper.Viper) *appSettings {
|
|||
|
||||
settings.setBypassContentEncodingInChunks(v.GetBool(cfgKludgeBypassContentEncodingCheckInChunks))
|
||||
settings.setClientCut(v.GetBool(cfgClientCut))
|
||||
settings.setMD5Enabled(v.GetBool(cfgMD5Enabled))
|
||||
|
||||
return settings
|
||||
}
|
||||
|
@ -196,6 +198,14 @@ func (s *appSettings) setClientCut(clientCut bool) {
|
|||
s.clientCut.Store(clientCut)
|
||||
}
|
||||
|
||||
func (s *appSettings) MD5Enabled() bool {
|
||||
return s.md5Enabled.Load()
|
||||
}
|
||||
|
||||
func (s *appSettings) setMD5Enabled(md5Enabled bool) {
|
||||
s.md5Enabled.Store(md5Enabled)
|
||||
}
|
||||
|
||||
func (a *App) initAPI(ctx context.Context) {
|
||||
a.initLayer(ctx)
|
||||
a.initHandler()
|
||||
|
@ -536,6 +546,7 @@ func (a *App) updateSettings() {
|
|||
a.settings.xmlDecoder.UseDefaultNamespaceForCompleteMultipart(a.cfg.GetBool(cfgKludgeUseDefaultXMLNSForCompleteMultipartUpload))
|
||||
a.settings.setBypassContentEncodingInChunks(a.cfg.GetBool(cfgKludgeBypassContentEncodingCheckInChunks))
|
||||
a.settings.setClientCut(a.cfg.GetBool(cfgClientCut))
|
||||
a.settings.setMD5Enabled(a.cfg.GetBool(cfgMD5Enabled))
|
||||
}
|
||||
|
||||
func (a *App) startServices() {
|
||||
|
@ -679,6 +690,7 @@ func (a *App) initHandler() {
|
|||
|
||||
cfg.CompleteMultipartKeepalive = a.cfg.GetDuration(cfgKludgeCompleteMultipartUploadKeepalive)
|
||||
cfg.Kludge = a.settings
|
||||
cfg.Features = a.settings
|
||||
|
||||
var err error
|
||||
a.api, err = handler.New(a.log, a.obj, a.nc, cfg)
|
||||
|
|
|
@ -160,6 +160,9 @@ const ( // Settings.
|
|||
// Runtime.
|
||||
cfgSoftMemoryLimit = "runtime.soft_memory_limit"
|
||||
|
||||
// Enable return MD5 checksum in ETag.
|
||||
cfgMD5Enabled = "features.md5.enabled"
|
||||
|
||||
// envPrefix is an environment variables prefix used for configuration.
|
||||
envPrefix = "S3_GW"
|
||||
)
|
||||
|
|
|
@ -148,3 +148,5 @@ S3_GW_TRACING_ENDPOINT="localhost:4318"
|
|||
S3_GW_TRACING_EXPORTER="otlp_grpc"
|
||||
|
||||
S3_GW_RUNTIME_SOFT_MEMORY_LIMIT=1073741824
|
||||
|
||||
S3_GW_FEATURES_MD5_ENABLED=false
|
||||
|
|
|
@ -174,3 +174,7 @@ kludge:
|
|||
|
||||
runtime:
|
||||
soft_memory_limit: 1gb
|
||||
|
||||
features:
|
||||
md5:
|
||||
enabled: false
|
||||
|
|
|
@ -186,6 +186,7 @@ There are some custom types used for brevity:
|
|||
| `resolve_bucket` | [Bucket name resolving configuration](#resolve_bucket-section) |
|
||||
| `kludge` | [Different kludge configuration](#kludge-section) |
|
||||
| `runtime` | [Runtime configuration](#runtime-section) |
|
||||
| `features` | [Features configuration](#features-section) |
|
||||
|
||||
### General section
|
||||
|
||||
|
@ -560,3 +561,16 @@ runtime:
|
|||
| Parameter | Type | SIGHUP reload | Default value | Description |
|
||||
|---------------------|--------|---------------|---------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
|
||||
| `soft_memory_limit` | `size` | yes | maxint64 | Soft memory limit for the runtime. Zero or no value stands for no limit. If `GOMEMLIMIT` environment variable is set, the value from the configuration file will be ignored. |
|
||||
|
||||
# `features` section
|
||||
Contains parameters for enabling features.
|
||||
|
||||
```yaml
|
||||
features:
|
||||
md5:
|
||||
enabled: false
|
||||
```
|
||||
|
||||
| Parameter | Type | SIGHUP reload | Default value | Description |
|
||||
|---------------|--------|---------------|---------------|----------------------------------------------------------------|
|
||||
| `md5.enabled` | `bool` | yes | false | Flag to enable return MD5 checksum in ETag headers and fields. |
|
||||
|
|
|
@ -75,6 +75,7 @@ const (
|
|||
ResolveBucket = "resolve bucket" // Info in ../../api/layer/layer.go
|
||||
CouldntDeleteCorsObject = "couldn't delete cors object" // Error in ../../api/layer/cors.go
|
||||
PutObject = "put object" // Debug in ../../api/layer/object.go
|
||||
FailedToDeleteObject = "failed to delete object" // Debug in ../../api/layer/object.go
|
||||
FailedToDiscardPutPayloadProbablyGoroutineLeaks = "failed to discard put payload, probably goroutine leaks" // Warn in ../../api/layer/object.go
|
||||
FailedToSubmitTaskToPool = "failed to submit task to pool" // Warn in ../../api/layer/object.go
|
||||
CouldNotFetchObjectMeta = "could not fetch object meta" // Warn in ../../api/layer/object.go
|
||||
|
|
|
@ -81,6 +81,7 @@ const (
|
|||
partNumberKV = "Number"
|
||||
sizeKV = "Size"
|
||||
etagKV = "ETag"
|
||||
md5KV = "MD5"
|
||||
|
||||
// keys for lock.
|
||||
isLockKV = "IsLock"
|
||||
|
@ -185,6 +186,7 @@ func newNodeVersionFromTreeNode(filePath string, treeNode *treeNode) *data.NodeV
|
|||
_, isDeleteMarker := treeNode.Get(isDeleteMarkerKV)
|
||||
_, isCombined := treeNode.Get(isCombinedKV)
|
||||
eTag, _ := treeNode.Get(etagKV)
|
||||
md5, _ := treeNode.Get(md5KV)
|
||||
|
||||
version := &data.NodeVersion{
|
||||
BaseNodeVersion: data.BaseNodeVersion{
|
||||
|
@ -193,6 +195,7 @@ func newNodeVersionFromTreeNode(filePath string, treeNode *treeNode) *data.NodeV
|
|||
OID: treeNode.ObjID,
|
||||
Timestamp: treeNode.TimeStamp,
|
||||
ETag: eTag,
|
||||
MD5: md5,
|
||||
Size: treeNode.Size,
|
||||
FilePath: filePath,
|
||||
},
|
||||
|
@ -302,6 +305,8 @@ func newPartInfo(node NodeResponse) (*data.PartInfo, error) {
|
|||
return nil, fmt.Errorf("invalid created timestamp: %w", err)
|
||||
}
|
||||
partInfo.Created = time.UnixMilli(utcMilli)
|
||||
case md5KV:
|
||||
partInfo.MD5 = value
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -560,7 +565,7 @@ func (c *Tree) GetVersions(ctx context.Context, bktInfo *data.BucketInfo, filepa
|
|||
}
|
||||
|
||||
func (c *Tree) GetLatestVersion(ctx context.Context, bktInfo *data.BucketInfo, objectName string) (*data.NodeVersion, error) {
|
||||
meta := []string{oidKV, isUnversionedKV, isDeleteMarkerKV, etagKV, sizeKV}
|
||||
meta := []string{oidKV, isUnversionedKV, isDeleteMarkerKV, etagKV, sizeKV, md5KV}
|
||||
path := pathFromName(objectName)
|
||||
|
||||
p := &GetNodesParams{
|
||||
|
@ -1006,6 +1011,7 @@ func (c *Tree) AddPart(ctx context.Context, bktInfo *data.BucketInfo, multipartN
|
|||
sizeKV: strconv.FormatUint(info.Size, 10),
|
||||
createdKV: strconv.FormatInt(info.Created.UTC().UnixMilli(), 10),
|
||||
etagKV: info.ETag,
|
||||
md5KV: info.MD5,
|
||||
}
|
||||
|
||||
for _, part := range parts {
|
||||
|
@ -1138,6 +1144,9 @@ func (c *Tree) addVersion(ctx context.Context, bktInfo *data.BucketInfo, treeID
|
|||
if len(version.ETag) > 0 {
|
||||
meta[etagKV] = version.ETag
|
||||
}
|
||||
if len(version.MD5) > 0 {
|
||||
meta[md5KV] = version.MD5
|
||||
}
|
||||
|
||||
if version.IsDeleteMarker() {
|
||||
meta[isDeleteMarkerKV] = "true"
|
||||
|
@ -1182,7 +1191,7 @@ func (c *Tree) clearOutdatedVersionInfo(ctx context.Context, bktInfo *data.Bucke
|
|||
}
|
||||
|
||||
func (c *Tree) getVersions(ctx context.Context, bktInfo *data.BucketInfo, treeID, filepath string, onlyUnversioned bool) ([]*data.NodeVersion, error) {
|
||||
keysToReturn := []string{oidKV, isUnversionedKV, isDeleteMarkerKV, etagKV, sizeKV}
|
||||
keysToReturn := []string{oidKV, isUnversionedKV, isDeleteMarkerKV, etagKV, sizeKV, md5KV}
|
||||
path := pathFromName(filepath)
|
||||
p := &GetNodesParams{
|
||||
BktInfo: bktInfo,
|
||||
|
|
Loading…
Reference in a new issue