frostfs-node/pkg/core/object/sender_classifier.go
Dmitrii Stepanov 0144117cc9 [] objectSvc: Add EC header APE check
Signed-off-by: Dmitrii Stepanov <d.stepanov@yadro.com>
2024-05-08 16:25:55 +03:00

161 lines
4 KiB
Go

package object
import (
"bytes"
"crypto/sha256"
"git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs"
core "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"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/acl"
cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/netmap"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/user"
"github.com/nspcc-dev/neo-go/pkg/crypto/keys"
"go.uber.org/zap"
)
type InnerRing interface {
InnerRingKeys() ([][]byte, error)
}
type SenderClassifier struct {
log *logger.Logger
innerRing InnerRing
netmap core.Source
}
func NewSenderClassifier(innerRing InnerRing, netmap core.Source, log *logger.Logger) SenderClassifier {
return SenderClassifier{
log: log,
innerRing: innerRing,
netmap: netmap,
}
}
type ClassifyResult struct {
Role acl.Role
Key []byte
}
func (c SenderClassifier) Classify(
ownerID *user.ID,
ownerKey *keys.PublicKey,
idCnr cid.ID,
cnr container.Container,
) (res *ClassifyResult, err error) {
ownerKeyInBytes := ownerKey.Bytes()
// TODO: #767 get owner from frostfs.id if present
// if request owner is the same as container owner, return RoleUser
if ownerID.Equals(cnr.Owner()) {
return &ClassifyResult{
Role: acl.RoleOwner,
Key: ownerKeyInBytes,
}, nil
}
return c.IsInnerRingOrContainerNode(ownerKeyInBytes, idCnr, cnr)
}
func (c SenderClassifier) IsInnerRingOrContainerNode(ownerKeyInBytes []byte, idCnr cid.ID, cnr container.Container) (*ClassifyResult, error) {
isInnerRingNode, err := c.isInnerRingKey(ownerKeyInBytes)
if err != nil {
// do not throw error, try best case matching
c.log.Debug(logs.V2CantCheckIfRequestFromInnerRing,
zap.String("error", err.Error()))
} else if isInnerRingNode {
return &ClassifyResult{
Role: acl.RoleInnerRing,
Key: ownerKeyInBytes,
}, nil
}
binCnr := make([]byte, sha256.Size)
idCnr.Encode(binCnr)
isContainerNode, err := c.isContainerKey(ownerKeyInBytes, binCnr, cnr)
if err != nil {
// error might happen if request has `RoleOther` key and placement
// is not possible for previous epoch, so
// do not throw error, try best case matching
c.log.Debug(logs.V2CantCheckIfRequestFromContainerNode,
zap.String("error", err.Error()))
} else if isContainerNode {
return &ClassifyResult{
Role: acl.RoleContainer,
Key: ownerKeyInBytes,
}, nil
}
// if none of above, return RoleOthers
return &ClassifyResult{
Role: acl.RoleOthers,
Key: ownerKeyInBytes,
}, nil
}
func (c SenderClassifier) isInnerRingKey(owner []byte) (bool, error) {
innerRingKeys, err := c.innerRing.InnerRingKeys()
if err != nil {
return false, err
}
// if request owner key in the inner ring list, return RoleSystem
for i := range innerRingKeys {
if bytes.Equal(innerRingKeys[i], owner) {
return true, nil
}
}
return false, nil
}
func (c SenderClassifier) isContainerKey(
owner, idCnr []byte,
cnr container.Container,
) (bool, error) {
nm, err := core.GetLatestNetworkMap(c.netmap) // first check current netmap
if err != nil {
return false, err
}
in, err := LookupKeyInContainer(nm, owner, idCnr, cnr)
if err != nil {
return false, err
} else if in {
return true, nil
}
// then check previous netmap, this can happen in-between epoch change
// when node migrates data from last epoch container
nm, err = core.GetPreviousNetworkMap(c.netmap)
if err != nil {
return false, err
}
return LookupKeyInContainer(nm, owner, idCnr, cnr)
}
func LookupKeyInContainer(
nm *netmap.NetMap,
pkey, idCnr []byte,
cnr container.Container,
) (bool, error) {
cnrVectors, err := nm.ContainerNodes(cnr.PlacementPolicy(), idCnr)
if err != nil {
return false, err
}
for i := range cnrVectors {
for j := range cnrVectors[i] {
if bytes.Equal(cnrVectors[i][j].PublicKey(), pkey) {
return true, nil
}
}
}
return false, nil
}