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

470 lines
12 KiB
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 || !token.AssertAuthKey(&key) {
return v.checkOwnerKey(ownerID, key)
}
if v.verifyTokenIssuer {
signerIsIROrContainerNode, err := v.isIROrContainerNode(obj, binKey)
if err != nil {
return err
}
if signerIsIROrContainerNode {
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) (bool, error) {
cnrID, containerIDSet := obj.ContainerID()
if !containerIDSet {
return false, errNilCID
}
cnrIDBin := make([]byte, sha256.Size)
cnrID.Encode(cnrIDBin)
cnr, err := v.containers.Get(cnrID)
if err != nil {
return false, 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 false, err
}
return res.Role == acl.RoleContainer || res.Role == acl.RoleInnerRing, 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
}
}