package object import ( "context" "crypto/ecdsa" "crypto/sha256" "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/container" "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/netmap" "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/util/logger" "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/acl" cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id" frostfsecdsa "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/crypto/ecdsa" objectSDK "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/user" ) // FormatValidator represents an object format validator. type FormatValidator struct { *cfg senderClassifier SenderClassifier } // FormatValidatorOption represents a FormatValidator constructor option. type FormatValidatorOption func(*cfg) type cfg struct { netState netmap.State e LockSource ir InnerRing netmap netmap.Source containers container.Source log *logger.Logger verifyTokenIssuer bool } // 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 errMissingSignature = errors.New("missing signature") 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, senderClassifier: NewSenderClassifier(cfg.ir, cfg.netmap, cfg.log), } } // 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 *objectSDK.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 := objectSDK.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 *objectSDK.Object) error { sig := obj.Signature() if sig == nil { return errMissingSignature } 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() ownerID := obj.OwnerID() if token == nil && obj.ECHeader() != nil { role, err := v.isIROrContainerNode(obj, binKey) if err != nil { return err } if role == acl.RoleContainer { // EC part could be restored or created by container node, so ownerID could not match object signature return nil } return v.checkOwnerKey(ownerID, key) } if token == nil || !token.AssertAuthKey(&key) { return v.checkOwnerKey(ownerID, key) } if v.verifyTokenIssuer { role, err := v.isIROrContainerNode(obj, binKey) if err != nil { return err } if role == acl.RoleContainer || role == acl.RoleInnerRing { return nil } if !token.Issuer().Equals(ownerID) { return fmt.Errorf("(%T) different token issuer and object owner identifiers %s/%s", v, token.Issuer(), ownerID) } return nil } return nil } func (v *FormatValidator) isIROrContainerNode(obj *objectSDK.Object, signerKey []byte) (acl.Role, error) { cnrID, containerIDSet := obj.ContainerID() if !containerIDSet { return acl.RoleOthers, errNilCID } cnrIDBin := make([]byte, sha256.Size) cnrID.Encode(cnrIDBin) cnr, err := v.containers.Get(cnrID) if err != nil { return acl.RoleOthers, fmt.Errorf("failed to get container (id=%s): %w", cnrID.EncodeToString(), err) } res, err := v.senderClassifier.IsInnerRingOrContainerNode(signerKey, cnrID, cnr.Value) if err != nil { return acl.RoleOthers, err } return res.Role, 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.TypeLock. type ContentMeta struct { typ objectSDK.Type objs []oid.ID } // Type returns object's type. func (i ContentMeta) Type() objectSDK.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; // - 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 *objectSDK.Object) (ContentMeta, error) { meta := ContentMeta{ typ: o.Type(), } switch o.Type() { case objectSDK.TypeTombstone: if err := v.fillAndValidateTombstoneMeta(o, &meta); err != nil { return ContentMeta{}, err } case objectSDK.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 *objectSDK.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 objectSDK.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) fillAndValidateTombstoneMeta(o *objectSDK.Object, meta *ContentMeta) error { if len(o.Payload()) == 0 { return fmt.Errorf("(%T) empty payload in tombstone", v) } tombstone := objectSDK.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 *objectSDK.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 *objectSDK.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 *objectSDK.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 *objectSDK.Object) error { if idOwner := obj.OwnerID(); idOwner.IsEmpty() { 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 } } // WithInnerRing return option to set Inner Ring source. func WithInnerRing(ir InnerRing) FormatValidatorOption { return func(c *cfg) { c.ir = ir } } // WithNetmapSource return option to set Netmap source. func WithNetmapSource(ns netmap.Source) FormatValidatorOption { return func(c *cfg) { c.netmap = ns } } // WithContainersSource return option to set Containers source. func WithContainersSource(cs container.Source) FormatValidatorOption { return func(c *cfg) { c.containers = cs } } // WithVerifySessionTokenIssuer return option to set verify session token issuer value. func WithVerifySessionTokenIssuer(verifySessionTokenIssuer bool) FormatValidatorOption { return func(c *cfg) { c.verifyTokenIssuer = verifySessionTokenIssuer } } // WithLogger return option to set logger. func WithLogger(l *logger.Logger) FormatValidatorOption { return func(c *cfg) { c.log = l } }