From 7de73f6b73b8be08c9151291f85d0e0cb6946494 Mon Sep 17 00:00:00 2001 From: Denis Kirillov Date: Fri, 25 Aug 2023 13:06:43 +0300 Subject: [PATCH] [#197] Disable homomorphic hash for PUT Disable TZ hash for PUT if it's disabled for container itself Signed-off-by: Denis Kirillov --- api/data/info.go | 15 ++++++++------- api/handler/handlers_test.go | 11 ++++++----- api/layer/container.go | 6 ++++-- api/layer/frostfs.go | 11 ++++++++++- api/layer/frostfs_mock.go | 10 +++++++--- api/layer/object.go | 1 + api/layer/versioning_test.go | 9 +++++---- internal/frostfs/authmate.go | 6 +++++- internal/frostfs/frostfs.go | 11 ++++++++--- 9 files changed, 54 insertions(+), 26 deletions(-) diff --git a/api/data/info.go b/api/data/info.go index 8530a7f..2949ccd 100644 --- a/api/data/info.go +++ b/api/data/info.go @@ -22,13 +22,14 @@ const ( type ( // BucketInfo stores basic bucket data. BucketInfo struct { - Name string // container name from system attribute - Zone string // container zone from system attribute - CID cid.ID - Owner user.ID - Created time.Time - LocationConstraint string - ObjectLockEnabled bool + Name string // container name from system attribute + Zone string // container zone from system attribute + CID cid.ID + Owner user.ID + Created time.Time + LocationConstraint string + ObjectLockEnabled bool + HomomorphicHashDisabled bool } // ObjectInfo holds S3 object data. diff --git a/api/handler/handlers_test.go b/api/handler/handlers_test.go index 2302303..b8c5227 100644 --- a/api/handler/handlers_test.go +++ b/api/handler/handlers_test.go @@ -213,7 +213,7 @@ func createTestBucket(hc *handlerContext, bktName string) *data.BucketInfo { } func createTestBucketWithLock(hc *handlerContext, bktName string, conf *data.ObjectLockConfiguration) *data.BucketInfo { - cnrID, err := hc.MockedPool().CreateContainer(hc.Context(), layer.PrmContainerCreate{ + res, err := hc.MockedPool().CreateContainer(hc.Context(), layer.PrmContainerCreate{ Creator: hc.owner, Name: bktName, AdditionalAttributes: [][2]string{{layer.AttributeLockEnabled, "true"}}, @@ -223,10 +223,11 @@ func createTestBucketWithLock(hc *handlerContext, bktName string, conf *data.Obj var ownerID user.ID bktInfo := &data.BucketInfo{ - CID: cnrID, - Name: bktName, - ObjectLockEnabled: true, - Owner: ownerID, + CID: res.ContainerID, + Name: bktName, + ObjectLockEnabled: true, + Owner: ownerID, + HomomorphicHashDisabled: res.HomomorphicHashDisabled, } sp := &layer.PutSettingsParams{ diff --git a/api/layer/container.go b/api/layer/container.go index e86dcb7..a9c9590 100644 --- a/api/layer/container.go +++ b/api/layer/container.go @@ -59,6 +59,7 @@ func (n *layer) containerInfo(ctx context.Context, idCnr cid.ID) (*data.BucketIn } info.Created = container.CreatedAt(cnr) info.LocationConstraint = cnr.Attribute(attributeLocationConstraint) + info.HomomorphicHashDisabled = container.IsHomomorphicHashingDisabled(cnr) attrLockEnabled := cnr.Attribute(AttributeLockEnabled) if len(attrLockEnabled) > 0 { @@ -122,7 +123,7 @@ func (n *layer) createContainer(ctx context.Context, p *CreateBucketParams) (*da }) } - idCnr, err := n.frostFS.CreateContainer(ctx, PrmContainerCreate{ + res, err := n.frostFS.CreateContainer(ctx, PrmContainerCreate{ Creator: bktInfo.Owner, Policy: p.Policy, Name: p.Name, @@ -134,7 +135,8 @@ func (n *layer) createContainer(ctx context.Context, p *CreateBucketParams) (*da return nil, fmt.Errorf("create container: %w", err) } - bktInfo.CID = idCnr + bktInfo.CID = res.ContainerID + bktInfo.HomomorphicHashDisabled = res.HomomorphicHashDisabled if err = n.setContainerEACLTable(ctx, bktInfo.CID, p.EACL, p.SessionEACL); err != nil { return nil, fmt.Errorf("set container eacl: %w", err) diff --git a/api/layer/frostfs.go b/api/layer/frostfs.go index 26f7352..01f8ece 100644 --- a/api/layer/frostfs.go +++ b/api/layer/frostfs.go @@ -43,6 +43,12 @@ type PrmContainerCreate struct { AdditionalAttributes [][2]string } +// ContainerCreateResult is a result parameter of FrostFS.CreateContainer operation. +type ContainerCreateResult struct { + ContainerID cid.ID + HomomorphicHashDisabled bool +} + // PrmAuth groups authentication parameters for the FrostFS operation. type PrmAuth struct { // Bearer token to be used for the operation. Overlaps PrivateKey. Optional. @@ -114,6 +120,9 @@ type PrmObjectCreate struct { // Enables client side object preparing. ClientCut bool + + // Disables using Tillich-Zémor hash for payload. + WithoutHomomorphicHash bool } // PrmObjectDelete groups parameters of FrostFS.DeleteObject operation. @@ -162,7 +171,7 @@ type FrostFS interface { // // It returns exactly one non-zero value. It returns any error encountered which // prevented the container from being created. - CreateContainer(context.Context, PrmContainerCreate) (cid.ID, error) + CreateContainer(context.Context, PrmContainerCreate) (*ContainerCreateResult, error) // Container reads a container from FrostFS by ID. // diff --git a/api/layer/frostfs_mock.go b/api/layer/frostfs_mock.go index 32eac88..e46f8b2 100644 --- a/api/layer/frostfs_mock.go +++ b/api/layer/frostfs_mock.go @@ -29,6 +29,10 @@ type FeatureSettingsMock struct { clientCut bool } +func (k *FeatureSettingsMock) BufferMaxSizeForPut() uint64 { + return 0 +} + func (k *FeatureSettingsMock) ClientCut() bool { return k.clientCut } @@ -114,7 +118,7 @@ func (t *TestFrostFS) ContainerID(name string) (cid.ID, error) { return cid.ID{}, fmt.Errorf("not found") } -func (t *TestFrostFS) CreateContainer(_ context.Context, prm PrmContainerCreate) (cid.ID, error) { +func (t *TestFrostFS) CreateContainer(_ context.Context, prm PrmContainerCreate) (*ContainerCreateResult, error) { var cnr container.Container cnr.Init() cnr.SetOwner(prm.Creator) @@ -141,14 +145,14 @@ func (t *TestFrostFS) CreateContainer(_ context.Context, prm PrmContainerCreate) b := make([]byte, 32) if _, err := io.ReadFull(rand.Reader, b); err != nil { - return cid.ID{}, err + return nil, err } var id cid.ID id.SetSHA256(sha256.Sum256(b)) t.containers[id.EncodeToString()] = &cnr - return id, nil + return &ContainerCreateResult{ContainerID: id}, nil } func (t *TestFrostFS) DeleteContainer(_ context.Context, cnrID cid.ID, _ *session.Container) error { diff --git a/api/layer/object.go b/api/layer/object.go index b9cde71..3f7dec0 100644 --- a/api/layer/object.go +++ b/api/layer/object.go @@ -460,6 +460,7 @@ func (n *layer) objectDelete(ctx context.Context, bktInfo *data.BucketInfo, idOb func (n *layer) objectPutAndHash(ctx context.Context, prm PrmObjectCreate, bktInfo *data.BucketInfo) (uint64, oid.ID, []byte, error) { n.prepareAuthParameters(ctx, &prm.PrmAuth, bktInfo.Owner) prm.ClientCut = n.features.ClientCut() + prm.WithoutHomomorphicHash = bktInfo.HomomorphicHashDisabled var size uint64 hash := sha256.New() prm.Payload = wrapReader(prm.Payload, 64*1024, func(buf []byte) { diff --git a/api/layer/versioning_test.go b/api/layer/versioning_test.go index 4476978..d59a3b9 100644 --- a/api/layer/versioning_test.go +++ b/api/layer/versioning_test.go @@ -153,7 +153,7 @@ func prepareContext(t *testing.T, cachesConfig ...*CachesConfig) *testContext { tp := NewTestFrostFS(key) bktName := "testbucket1" - bktID, err := tp.CreateContainer(ctx, PrmContainerCreate{ + res, err := tp.CreateContainer(ctx, PrmContainerCreate{ Name: bktName, }) require.NoError(t, err) @@ -177,9 +177,10 @@ func prepareContext(t *testing.T, cachesConfig ...*CachesConfig) *testContext { ctx: ctx, layer: NewLayer(logger, tp, layerCfg), bktInfo: &data.BucketInfo{ - Name: bktName, - Owner: owner, - CID: bktID, + Name: bktName, + Owner: owner, + CID: res.ContainerID, + HomomorphicHashDisabled: res.HomomorphicHashDisabled, }, obj: "obj1", t: t, diff --git a/internal/frostfs/authmate.go b/internal/frostfs/authmate.go index a53babb..2e9e423 100644 --- a/internal/frostfs/authmate.go +++ b/internal/frostfs/authmate.go @@ -57,12 +57,16 @@ func (x *AuthmateFrostFS) CreateContainer(ctx context.Context, prm authmate.PrmC basicACL.AllowOp(acl.OpObjectHead, acl.RoleOthers) basicACL.AllowOp(acl.OpObjectSearch, acl.RoleOthers) - return x.frostFS.CreateContainer(ctx, layer.PrmContainerCreate{ + res, err := x.frostFS.CreateContainer(ctx, layer.PrmContainerCreate{ Creator: prm.Owner, Policy: prm.Policy, Name: prm.FriendlyName, BasicACL: basicACL, }) + if err != nil { + return cid.ID{}, err + } + return res.ContainerID, nil } // GetCredsPayload implements authmate.FrostFS interface method. diff --git a/internal/frostfs/frostfs.go b/internal/frostfs/frostfs.go index ba66be8..d410e7b 100644 --- a/internal/frostfs/frostfs.go +++ b/internal/frostfs/frostfs.go @@ -106,7 +106,7 @@ var basicACLZero acl.Basic // CreateContainer implements frostfs.FrostFS interface method. // // If prm.BasicACL is zero, 'eacl-public-read-write' is used. -func (x *FrostFS) CreateContainer(ctx context.Context, prm layer.PrmContainerCreate) (cid.ID, error) { +func (x *FrostFS) CreateContainer(ctx context.Context, prm layer.PrmContainerCreate) (*layer.ContainerCreateResult, error) { if prm.BasicACL == basicACLZero { prm.BasicACL = acl.PublicRWExtended } @@ -137,7 +137,7 @@ func (x *FrostFS) CreateContainer(ctx context.Context, prm layer.PrmContainerCre err := pool.SyncContainerWithNetwork(ctx, &cnr, x.pool) if err != nil { - return cid.ID{}, handleObjectError("sync container with the network state", err) + return nil, handleObjectError("sync container with the network state", err) } prmPut := pool.PrmContainerPut{ @@ -150,7 +150,10 @@ func (x *FrostFS) CreateContainer(ctx context.Context, prm layer.PrmContainerCre // send request to save the container idCnr, err := x.pool.PutContainer(ctx, prmPut) - return idCnr, handleObjectError("save container via connection pool", err) + return &layer.ContainerCreateResult{ + ContainerID: idCnr, + HomomorphicHashDisabled: container.IsHomomorphicHashingDisabled(cnr), + }, handleObjectError("save container via connection pool", err) } // UserContainers implements frostfs.FrostFS interface method. @@ -244,6 +247,8 @@ func (x *FrostFS) CreateObject(ctx context.Context, prm layer.PrmObjectCreate) ( prmPut.SetPayload(prm.Payload) prmPut.SetCopiesNumberVector(prm.CopiesNumber) prmPut.SetClientCut(prm.ClientCut) + prmPut.WithoutHomomorphicHash(prm.WithoutHomomorphicHash) + prmPut.SetBufferMaxSize(prm.BufferMaxSize) if prm.BearerToken != nil { prmPut.UseBearer(*prm.BearerToken)