Dmitrii Stepanov
d62c6e4ce6
Add tracing spans for PUT requests. Add tracing spans for DELETE requests. Add tracing spans for SELECT requests. Signed-off-by: Dmitrii Stepanov <d.stepanov@yadro.com>
426 lines
10 KiB
Go
426 lines
10 KiB
Go
package object
|
|
|
|
import (
|
|
"context"
|
|
"crypto/ecdsa"
|
|
"errors"
|
|
"fmt"
|
|
"strconv"
|
|
|
|
objectV2 "git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/object"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/refs"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/netmap"
|
|
cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id"
|
|
frostfsecdsa "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/crypto/ecdsa"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
|
|
oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/storagegroup"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/user"
|
|
)
|
|
|
|
// FormatValidator represents an object format validator.
|
|
type FormatValidator struct {
|
|
*cfg
|
|
}
|
|
|
|
// FormatValidatorOption represents a FormatValidator constructor option.
|
|
type FormatValidatorOption func(*cfg)
|
|
|
|
type cfg struct {
|
|
netState netmap.State
|
|
e LockSource
|
|
}
|
|
|
|
// DeleteHandler is an interface of delete queue processor.
|
|
type DeleteHandler interface {
|
|
// DeleteObjects places objects to a removal queue.
|
|
//
|
|
// Returns apistatus.LockNonRegularObject if at least one object
|
|
// is locked.
|
|
DeleteObjects(oid.Address, ...oid.Address) error
|
|
}
|
|
|
|
// LockSource is a source of lock relations between the objects.
|
|
type LockSource interface {
|
|
// IsLocked must clarify object's lock status.
|
|
IsLocked(ctx context.Context, address oid.Address) (bool, error)
|
|
}
|
|
|
|
// Locker is an object lock storage interface.
|
|
type Locker interface {
|
|
// Lock list of objects as locked by locker in the specified container.
|
|
//
|
|
// Returns apistatus.LockNonRegularObject if at least object in locked
|
|
// list is irregular (not type of REGULAR).
|
|
Lock(idCnr cid.ID, locker oid.ID, locked []oid.ID) error
|
|
}
|
|
|
|
var errNilObject = errors.New("object is nil")
|
|
|
|
var errNilID = errors.New("missing identifier")
|
|
|
|
var errNilCID = errors.New("missing container identifier")
|
|
|
|
var errNoExpirationEpoch = errors.New("missing expiration epoch attribute")
|
|
|
|
var errTombstoneExpiration = errors.New("tombstone body and header contain different expiration values")
|
|
|
|
var errEmptySGMembers = errors.New("storage group with empty members list")
|
|
|
|
func defaultCfg() *cfg {
|
|
return new(cfg)
|
|
}
|
|
|
|
// NewFormatValidator creates, initializes and returns FormatValidator instance.
|
|
func NewFormatValidator(opts ...FormatValidatorOption) *FormatValidator {
|
|
cfg := defaultCfg()
|
|
|
|
for i := range opts {
|
|
opts[i](cfg)
|
|
}
|
|
|
|
return &FormatValidator{
|
|
cfg: cfg,
|
|
}
|
|
}
|
|
|
|
// Validate validates object format.
|
|
//
|
|
// Does not validate payload checksum and content.
|
|
// If unprepared is true, only fields set by user are validated.
|
|
//
|
|
// Returns nil error if the object has valid structure.
|
|
func (v *FormatValidator) Validate(ctx context.Context, obj *object.Object, unprepared bool) error {
|
|
if obj == nil {
|
|
return errNilObject
|
|
}
|
|
|
|
_, idSet := obj.ID()
|
|
if !unprepared && !idSet {
|
|
return errNilID
|
|
}
|
|
|
|
_, cnrSet := obj.ContainerID()
|
|
if !cnrSet {
|
|
return errNilCID
|
|
}
|
|
|
|
if err := v.checkOwner(obj); err != nil {
|
|
return err
|
|
}
|
|
|
|
if err := v.checkAttributes(obj); err != nil {
|
|
return fmt.Errorf("invalid attributes: %w", err)
|
|
}
|
|
|
|
if !unprepared {
|
|
if err := v.validateSignatureKey(obj); err != nil {
|
|
return fmt.Errorf("(%T) could not validate signature key: %w", v, err)
|
|
}
|
|
|
|
if err := v.checkExpiration(ctx, obj); err != nil {
|
|
return fmt.Errorf("object did not pass expiration check: %w", err)
|
|
}
|
|
|
|
if err := object.CheckHeaderVerificationFields(obj); err != nil {
|
|
return fmt.Errorf("(%T) could not validate header fields: %w", v, err)
|
|
}
|
|
}
|
|
|
|
if obj = obj.Parent(); obj != nil {
|
|
// Parent object already exists.
|
|
return v.Validate(ctx, obj, false)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (v *FormatValidator) validateSignatureKey(obj *object.Object) error {
|
|
// FIXME(@cthulhu-rider): temp solution, see neofs-sdk-go#233
|
|
sig := obj.Signature()
|
|
if sig == nil {
|
|
// TODO(@cthulhu-rider): #1387 use "const" error
|
|
return errors.New("missing signature")
|
|
}
|
|
|
|
var sigV2 refs.Signature
|
|
sig.WriteToV2(&sigV2)
|
|
|
|
binKey := sigV2.GetKey()
|
|
|
|
var key frostfsecdsa.PublicKey
|
|
|
|
err := key.Decode(binKey)
|
|
if err != nil {
|
|
return fmt.Errorf("decode public key: %w", err)
|
|
}
|
|
|
|
token := obj.SessionToken()
|
|
|
|
if token == nil || !token.AssertAuthKey(&key) {
|
|
return v.checkOwnerKey(*obj.OwnerID(), key)
|
|
}
|
|
|
|
// FIXME: #1159 perform token verification
|
|
|
|
return nil
|
|
}
|
|
|
|
func (v *FormatValidator) checkOwnerKey(id user.ID, key frostfsecdsa.PublicKey) error {
|
|
var id2 user.ID
|
|
user.IDFromKey(&id2, (ecdsa.PublicKey)(key))
|
|
|
|
if !id.Equals(id2) {
|
|
return fmt.Errorf("(%T) different owner identifiers %s/%s", v, id, id2)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// ContentMeta describes FrostFS meta information that brings object's payload if the object
|
|
// is one of:
|
|
// - object.TypeTombstone;
|
|
// - object.TypeStorageGroup;
|
|
// - object.TypeLock.
|
|
type ContentMeta struct {
|
|
typ object.Type
|
|
|
|
objs []oid.ID
|
|
}
|
|
|
|
// Type returns object's type.
|
|
func (i ContentMeta) Type() object.Type {
|
|
return i.typ
|
|
}
|
|
|
|
// Objects returns objects that the original object's payload affects:
|
|
// - inhumed objects, if the original object is a Tombstone;
|
|
// - locked objects, if the original object is a Lock;
|
|
// - members of a storage group, if the original object is a Storage group;
|
|
// - nil, if the original object is a Regular object.
|
|
func (i ContentMeta) Objects() []oid.ID {
|
|
return i.objs
|
|
}
|
|
|
|
// ValidateContent validates payload content according to the object type.
|
|
func (v *FormatValidator) ValidateContent(o *object.Object) (ContentMeta, error) {
|
|
meta := ContentMeta{
|
|
typ: o.Type(),
|
|
}
|
|
|
|
switch o.Type() {
|
|
case object.TypeTombstone:
|
|
if err := v.fillAndValidateTombstoneMeta(o, &meta); err != nil {
|
|
return ContentMeta{}, err
|
|
}
|
|
case object.TypeStorageGroup:
|
|
if err := v.fillAndValidateStorageGroupMeta(o, &meta); err != nil {
|
|
return ContentMeta{}, err
|
|
}
|
|
case object.TypeLock:
|
|
if err := v.fillAndValidateLockMeta(o, &meta); err != nil {
|
|
return ContentMeta{}, err
|
|
}
|
|
default:
|
|
// ignore all other object types, they do not need payload formatting
|
|
}
|
|
|
|
return meta, nil
|
|
}
|
|
|
|
func (v *FormatValidator) fillAndValidateLockMeta(o *object.Object, meta *ContentMeta) error {
|
|
if len(o.Payload()) == 0 {
|
|
return errors.New("empty payload in lock")
|
|
}
|
|
|
|
if _, ok := o.ContainerID(); !ok {
|
|
return errors.New("missing container")
|
|
}
|
|
|
|
if _, ok := o.ID(); !ok {
|
|
return errors.New("missing ID")
|
|
}
|
|
// check that LOCK object has correct expiration epoch
|
|
lockExp, err := expirationEpochAttribute(o)
|
|
if err != nil {
|
|
return fmt.Errorf("lock object expiration epoch: %w", err)
|
|
}
|
|
|
|
if currEpoch := v.netState.CurrentEpoch(); lockExp < currEpoch {
|
|
return fmt.Errorf("lock object expiration: %d; current: %d", lockExp, currEpoch)
|
|
}
|
|
|
|
var lock object.Lock
|
|
|
|
if err = lock.Unmarshal(o.Payload()); err != nil {
|
|
return fmt.Errorf("decode lock payload: %w", err)
|
|
}
|
|
|
|
num := lock.NumberOfMembers()
|
|
if num == 0 {
|
|
return errors.New("missing locked members")
|
|
}
|
|
|
|
meta.objs = make([]oid.ID, num)
|
|
lock.ReadMembers(meta.objs)
|
|
return nil
|
|
}
|
|
|
|
func (v *FormatValidator) fillAndValidateStorageGroupMeta(o *object.Object, meta *ContentMeta) error {
|
|
if len(o.Payload()) == 0 {
|
|
return fmt.Errorf("(%T) empty payload in storage group", v)
|
|
}
|
|
|
|
var sg storagegroup.StorageGroup
|
|
|
|
if err := sg.Unmarshal(o.Payload()); err != nil {
|
|
return fmt.Errorf("(%T) could not unmarshal storage group content: %w", v, err)
|
|
}
|
|
|
|
mm := sg.Members()
|
|
meta.objs = mm
|
|
|
|
lenMM := len(mm)
|
|
if lenMM == 0 {
|
|
return errEmptySGMembers
|
|
}
|
|
|
|
uniqueFilter := make(map[oid.ID]struct{}, lenMM)
|
|
|
|
for i := 0; i < lenMM; i++ {
|
|
if _, alreadySeen := uniqueFilter[mm[i]]; alreadySeen {
|
|
return fmt.Errorf("storage group contains non-unique member: %s", mm[i])
|
|
}
|
|
|
|
uniqueFilter[mm[i]] = struct{}{}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (v *FormatValidator) fillAndValidateTombstoneMeta(o *object.Object, meta *ContentMeta) error {
|
|
if len(o.Payload()) == 0 {
|
|
return fmt.Errorf("(%T) empty payload in tombstone", v)
|
|
}
|
|
|
|
tombstone := object.NewTombstone()
|
|
|
|
if err := tombstone.Unmarshal(o.Payload()); err != nil {
|
|
return fmt.Errorf("(%T) could not unmarshal tombstone content: %w", v, err)
|
|
}
|
|
// check if the tombstone has the same expiration in the body and the header
|
|
exp, err := expirationEpochAttribute(o)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if exp != tombstone.ExpirationEpoch() {
|
|
return errTombstoneExpiration
|
|
}
|
|
|
|
// mark all objects from the tombstone body as removed in the storage engine
|
|
if _, ok := o.ContainerID(); !ok {
|
|
return errors.New("missing container ID")
|
|
}
|
|
|
|
meta.objs = tombstone.Members()
|
|
return nil
|
|
}
|
|
|
|
var errExpired = errors.New("object has expired")
|
|
|
|
func (v *FormatValidator) checkExpiration(ctx context.Context, obj *object.Object) error {
|
|
exp, err := expirationEpochAttribute(obj)
|
|
if err != nil {
|
|
if errors.Is(err, errNoExpirationEpoch) {
|
|
return nil // objects without expiration attribute are valid
|
|
}
|
|
|
|
return err
|
|
}
|
|
|
|
if exp < v.netState.CurrentEpoch() {
|
|
// an object could be expired but locked;
|
|
// put such an object is a correct operation
|
|
|
|
cID, _ := obj.ContainerID()
|
|
oID, _ := obj.ID()
|
|
|
|
var addr oid.Address
|
|
addr.SetContainer(cID)
|
|
addr.SetObject(oID)
|
|
|
|
locked, err := v.e.IsLocked(ctx, addr)
|
|
if err != nil {
|
|
return fmt.Errorf("locking status check for an expired object: %w", err)
|
|
}
|
|
|
|
if !locked {
|
|
return errExpired
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func expirationEpochAttribute(obj *object.Object) (uint64, error) {
|
|
for _, a := range obj.Attributes() {
|
|
if a.Key() != objectV2.SysAttributeExpEpoch && a.Key() != objectV2.SysAttributeExpEpochNeoFS {
|
|
continue
|
|
}
|
|
|
|
return strconv.ParseUint(a.Value(), 10, 64)
|
|
}
|
|
|
|
return 0, errNoExpirationEpoch
|
|
}
|
|
|
|
var (
|
|
errDuplAttr = errors.New("duplication of attributes detected")
|
|
errEmptyAttrVal = errors.New("empty attribute value")
|
|
)
|
|
|
|
func (v *FormatValidator) checkAttributes(obj *object.Object) error {
|
|
as := obj.Attributes()
|
|
|
|
mUnique := make(map[string]struct{}, len(as))
|
|
|
|
for _, a := range as {
|
|
key := a.Key()
|
|
|
|
if _, was := mUnique[key]; was {
|
|
return errDuplAttr
|
|
}
|
|
|
|
if a.Value() == "" {
|
|
return errEmptyAttrVal
|
|
}
|
|
|
|
mUnique[key] = struct{}{}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
var errIncorrectOwner = errors.New("incorrect object owner")
|
|
|
|
func (v *FormatValidator) checkOwner(obj *object.Object) error {
|
|
if idOwner := obj.OwnerID(); idOwner == nil || len(idOwner.WalletBytes()) == 0 {
|
|
return errIncorrectOwner
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// WithNetState returns options to set the network state interface.
|
|
func WithNetState(netState netmap.State) FormatValidatorOption {
|
|
return func(c *cfg) {
|
|
c.netState = netState
|
|
}
|
|
}
|
|
|
|
// WithLockSource return option to set the Storage Engine.
|
|
func WithLockSource(e LockSource) FormatValidatorOption {
|
|
return func(c *cfg) {
|
|
c.e = e
|
|
}
|
|
}
|