achuprov/frostfs-node
1
0
Fork 0
forked from TrueCloudLab/frostfs-node
Code Pull requests Activity

[#264] object/get: Check parent address in linking/last child

Browse source 
Signed-off-by: Leonard Lyubich <leonard@nspcc.ru>
This commit is contained in:
Leonard Lyubich 2020-12-18 14:48:28 +03:00 committed by Leonard Lyubich
parent ec21040542
commit 66f9532857
4 changed files with 35 additions and 19 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

BIN
go.sum
View file

Binary file not shown.

19
pkg/services/object/get/assemble.go
View file

@ -26,7 +26,7 @@ func (exec *execCtx) assemble() {
if len(children) > 0 {
if exec.ctxRange() == nil {
if ok := exec.writeCollectedHeader(); ok {
exec.overtakePayloadDirectly(children, nil)
exec.overtakePayloadDirectly(children, nil, true)
}
} else {
// TODO: choose one-by-one restoring algorithm according to size
@ -59,7 +59,7 @@ func (exec *execCtx) initFromChild(id *objectSDK.ID) (prev *objectSDK.ID, childr
log.Debug("starting assembling from child")
child, ok := exec.getChild(id, nil)
child, ok := exec.getChild(id, nil, true)
if !ok {
return
}
@ -70,15 +70,6 @@ func (exec *execCtx) initFromChild(id *objectSDK.ID) (prev *objectSDK.ID, childr
log.Debug("received child with empty parent")
return
} else if !equalAddresses(par.Address(), exec.address()) {
exec.status = statusUndefined
log.Debug("parent address in child object differs",
zap.Stringer("expected", exec.address()),
zap.Stringer("received", par.Address()),
)
return
}
@ -122,7 +113,7 @@ func (exec *execCtx) initFromChild(id *objectSDK.ID) (prev *objectSDK.ID, childr
return child.PreviousID(), child.Children()
}
func (exec *execCtx) overtakePayloadDirectly(children []*objectSDK.ID, rngs []*objectSDK.Range) {
func (exec *execCtx) overtakePayloadDirectly(children []*objectSDK.ID, rngs []*objectSDK.Range, checkRight bool) {
withRng := len(rngs) > 0
for i := range children {
@ -131,7 +122,7 @@ func (exec *execCtx) overtakePayloadDirectly(children []*objectSDK.ID, rngs []*o
r = rngs[i]
}
child, ok := exec.getChild(children[i], r)
child, ok := exec.getChild(children[i], r, !withRng && checkRight)
if !ok {
return
}
@ -162,7 +153,7 @@ func (exec *execCtx) overtakePayloadInReverse(prev *objectSDK.ID) bool {
}
}
exec.overtakePayloadDirectly(chain, rngs)
exec.overtakePayloadDirectly(chain, rngs, false)
exec.status = statusOK
exec.err = nil

34
pkg/services/object/get/exec.go
View file

@ -3,6 +3,7 @@ package getsvc
import (
"context"
"crypto/ecdsa"
"errors"
"github.com/nspcc-dev/neofs-api-go/pkg/client"
"github.com/nspcc-dev/neofs-api-go/pkg/container"
@ -95,6 +96,11 @@ func (exec execCtx) address() *objectSDK.Address {
return exec.prm.Address()
}
func (exec execCtx) isChild(obj *object.Object) bool {
par := obj.GetParent()
return par != nil && equalAddresses(exec.address(), par.Address())
}
func (exec execCtx) key() *ecdsa.PrivateKey {
return exec.prm.common.PrivateKey()
}
@ -147,7 +153,7 @@ func (exec *execCtx) generateTraverser(addr *objectSDK.Address) (*placement.Trav
}
}
func (exec *execCtx) getChild(id *objectSDK.ID, rng *objectSDK.Range) (*object.Object, bool) {
func (exec *execCtx) getChild(id *objectSDK.ID, rng *objectSDK.Range, withHdr bool) (*object.Object, bool) {
w := NewSimpleObjectWriter()
p := exec.prm
@ -163,7 +169,17 @@ func (exec *execCtx) getChild(id *objectSDK.ID, rng *objectSDK.Range) (*object.O
exec.statusError = exec.svc.get(exec.context(), p.commonPrm, withPayloadRange(rng))
return w.Object(), exec.status == statusOK
child := w.Object()
ok := exec.status == statusOK
if ok && withHdr && !exec.isChild(child) {
exec.status = statusUndefined
exec.err = errors.New("wrong child header")
exec.log.Debug("parent address in child object differs")
}
return child, ok
}
func (exec *execCtx) headChild(id *objectSDK.ID) (*object.Object, bool) {
@ -196,10 +212,18 @@ func (exec *execCtx) headChild(id *objectSDK.ID) (*object.Object, bool) {
return nil, false
case err == nil:
exec.status = statusOK
exec.err = nil
child := w.Object()
return w.Object(), true
if child.ParentID() != nil && !exec.isChild(child) {
exec.status = statusUndefined
exec.log.Debug("parent address in child object differs")
} else {
exec.status = statusOK
exec.err = nil
}
return child, true
}
}

1
pkg/services/object/get/get_test.go
View file

@ -826,6 +826,7 @@ func TestGetRemoteSmall(t *testing.T) {
children, childIDs, payload := generateChain(2, cid)
srcObj.SetPayload(payload)
srcObj.SetPayloadSize(uint64(len(payload)))
children[len(children)-1].SetParent(srcObj.Object().SDK())
linkAddr := objectSDK.NewAddress()
linkAddr.SetContainerID(cid)