frostfs-node/pkg/core/object/fmt.go

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
	}
}