frostfs-node/pkg/services/object/ape/request.go

package ape

import (
	"context"
	"crypto/sha256"
	"errors"
	"fmt"
	"net"
	"strconv"

	objectV2 "git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/object"
	aperequest "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/ape/request"
	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/netmap"
	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/object"
	apistatus "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client/status"
	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/acl"
	cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id"
	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"
	commonschema "git.frostfs.info/TrueCloudLab/policy-engine/schema/common"
	nativeschema "git.frostfs.info/TrueCloudLab/policy-engine/schema/native"
	"github.com/nspcc-dev/neo-go/pkg/crypto/keys"
	"google.golang.org/grpc/peer"
)

var defaultRequest = aperequest.Request{}

var errECMissingParentObjectID = errors.New("missing EC parent object ID")

func nativeSchemaRole(role acl.Role) string {
	switch role {
	case acl.RoleOwner:
		return nativeschema.PropertyValueContainerRoleOwner
	case acl.RoleContainer:
		return nativeschema.PropertyValueContainerRoleContainer
	case acl.RoleInnerRing:
		return nativeschema.PropertyValueContainerRoleIR
	case acl.RoleOthers:
		return nativeschema.PropertyValueContainerRoleOthers
	default:
		return ""
	}
}

func resourceName(cid cid.ID, oid *oid.ID, namespace string) string {
	if namespace == "root" || namespace == "" {
		if oid != nil {
			return fmt.Sprintf(nativeschema.ResourceFormatRootContainerObject, cid.EncodeToString(), oid.EncodeToString())
		}
		return fmt.Sprintf(nativeschema.ResourceFormatRootContainerObjects, cid.EncodeToString())
	}
	if oid != nil {
		return fmt.Sprintf(nativeschema.ResourceFormatNamespaceContainerObject, namespace, cid.EncodeToString(), oid.EncodeToString())
	}
	return fmt.Sprintf(nativeschema.ResourceFormatNamespaceContainerObjects, namespace, cid.EncodeToString())
}

// objectProperties collects object properties from address parameters and a header if it is passed.
func objectProperties(cnr cid.ID, oid *oid.ID, cnrOwner user.ID, header *objectV2.Header) map[string]string {
	objectProps := map[string]string{
		nativeschema.PropertyKeyObjectContainerID: cnr.EncodeToString(),
	}

	objectProps[nativeschema.PropertyKeyContainerOwnerID] = cnrOwner.EncodeToString()

	if oid != nil {
		objectProps[nativeschema.PropertyKeyObjectID] = oid.String()
	}

	if header == nil {
		return objectProps
	}

	objV2 := new(objectV2.Object)
	objV2.SetHeader(header)
	objSDK := objectSDK.NewFromV2(objV2)

	objectProps[nativeschema.PropertyKeyObjectVersion] = objSDK.Version().String()
	objectProps[nativeschema.PropertyKeyObjectOwnerID] = objSDK.OwnerID().EncodeToString()
	objectProps[nativeschema.PropertyKeyObjectCreationEpoch] = strconv.Itoa(int(objSDK.CreationEpoch()))
	objectProps[nativeschema.PropertyKeyObjectPayloadLength] = strconv.Itoa(int(objSDK.PayloadSize()))
	objectProps[nativeschema.PropertyKeyObjectType] = objSDK.Type().String()

	pcs, isSet := objSDK.PayloadChecksum()
	if isSet {
		objectProps[nativeschema.PropertyKeyObjectPayloadHash] = pcs.String()
	}
	hcs, isSet := objSDK.PayloadHomomorphicHash()
	if isSet {
		objectProps[nativeschema.PropertyKeyObjectHomomorphicHash] = hcs.String()
	}

	for _, attr := range header.GetAttributes() {
		objectProps[attr.GetKey()] = attr.GetValue()
	}

	return objectProps
}

// newAPERequest creates an APE request to be passed to a chain router. It collects resource properties from
// header provided by headerProvider. If it cannot be found in headerProvider, then properties are
// initialized from header given in prm (if it is set). Otherwise, just CID and OID are set to properties.
func (c *checkerImpl) newAPERequest(ctx context.Context, prm Prm) (aperequest.Request, error) {
	switch prm.Method {
	case nativeschema.MethodGetObject,
		nativeschema.MethodHeadObject,
		nativeschema.MethodRangeObject,
		nativeschema.MethodHashObject,
		nativeschema.MethodDeleteObject,
		nativeschema.MethodPatchObject:
		if prm.Object == nil {
			return defaultRequest, fmt.Errorf("method %s: %w", prm.Method, errMissingOID)
		}
	case nativeschema.MethodSearchObject, nativeschema.MethodPutObject:
	default:
		return defaultRequest, fmt.Errorf("unknown method: %s", prm.Method)
	}

	var header *objectV2.Header
	if prm.Header != nil {
		header = prm.Header
	} else if prm.Object != nil {
		headerObjSDK, err := c.headerProvider.GetHeader(ctx, prm.Container, *prm.Object, true)
		if err == nil {
			header = headerObjSDK.ToV2().GetHeader()
		}
	}
	header, err := c.fillHeaderWithECParent(ctx, prm, header)
	if err != nil {
		return defaultRequest, fmt.Errorf("get EC parent header: %w", err)
	}
	reqProps := map[string]string{
		nativeschema.PropertyKeyActorPublicKey: prm.SenderKey,
		nativeschema.PropertyKeyActorRole:      prm.Role,
	}

	for _, xhead := range prm.XHeaders {
		xheadKey := fmt.Sprintf(commonschema.PropertyKeyFrostFSXHeader, xhead.GetKey())
		reqProps[xheadKey] = xhead.GetValue()
	}

	reqProps, err = c.fillWithUserClaimTags(reqProps, prm)
	if err != nil {
		return defaultRequest, err
	}

	if p, ok := peer.FromContext(ctx); ok {
		if tcpAddr, ok := p.Addr.(*net.TCPAddr); ok {
			reqProps[commonschema.PropertyKeyFrostFSSourceIP] = tcpAddr.IP.String()
		}
	}

	return aperequest.NewRequest(
		prm.Method,
		aperequest.NewResource(
			resourceName(prm.Container, prm.Object, prm.Namespace),
			objectProperties(prm.Container, prm.Object, prm.ContainerOwner, header),
		),
		reqProps,
	), nil
}

func (c *checkerImpl) fillHeaderWithECParent(ctx context.Context, prm Prm, header *objectV2.Header) (*objectV2.Header, error) {
	if header == nil {
		return header, nil
	}
	if header.GetEC() == nil {
		return header, nil
	}
	parentObjRefID := header.GetEC().Parent
	if parentObjRefID == nil {
		return nil, errECMissingParentObjectID
	}
	var parentObjID oid.ID
	if err := parentObjID.ReadFromV2(*parentObjRefID); err != nil {
		return nil, fmt.Errorf("EC parent object ID format error: %w", err)
	}
	// only container node have access to collect parent object
	contNode, err := c.currentNodeIsContainerNode(prm.Container)
	if err != nil {
		return nil, fmt.Errorf("check container node status: %w", err)
	}
	if !contNode {
		return header, nil
	}
	parentObj, err := c.headerProvider.GetHeader(ctx, prm.Container, parentObjID, false)
	if err != nil {
		if isLogicalError(err) {
			return header, nil
		}
		return nil, fmt.Errorf("EC parent header request: %w", err)
	}
	return parentObj.ToV2().GetHeader(), nil
}

func isLogicalError(err error) bool {
	var errObjRemoved *apistatus.ObjectAlreadyRemoved
	var errObjNotFound *apistatus.ObjectNotFound
	return errors.As(err, &errObjRemoved) || errors.As(err, &errObjNotFound)
}

func (c *checkerImpl) currentNodeIsContainerNode(cnrID cid.ID) (bool, error) {
	cnr, err := c.cnrSource.Get(cnrID)
	if err != nil {
		return false, err
	}

	nm, err := netmap.GetLatestNetworkMap(c.nm)
	if err != nil {
		return false, err
	}
	idCnr := make([]byte, sha256.Size)
	cnrID.Encode(idCnr)

	in, err := object.LookupKeyInContainer(nm, c.nodePK, idCnr, cnr.Value)
	if err != nil {
		return false, err
	} else if in {
		return true, nil
	}

	nm, err = netmap.GetPreviousNetworkMap(c.nm)
	if err != nil {
		return false, err
	}

	return object.LookupKeyInContainer(nm, c.nodePK, idCnr, cnr.Value)
}

// fillWithUserClaimTags fills ape request properties with user claim tags getting them from frostfsid contract by actor public key.
func (c *checkerImpl) fillWithUserClaimTags(reqProps map[string]string, prm Prm) (map[string]string, error) {
	if reqProps == nil {
		reqProps = make(map[string]string)
	}
	pk, err := keys.NewPublicKeyFromString(prm.SenderKey)
	if err != nil {
		return nil, err
	}
	props, err := aperequest.FormFrostfsIDRequestProperties(c.frostFSIDClient, pk)
	if err != nil {
		return reqProps, err
	}
	for propertyName, properyValue := range props {
		reqProps[propertyName] = properyValue
	}
	return reqProps, nil
}