frostfs-node/pkg/services/object/get/v2/head_forwarder.go

package getsvc

import (
	"context"
	"crypto/ecdsa"
	"errors"
	"fmt"
	"sync"

	objectV2 "git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/object"
	"git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/pkg/tracing"
	"git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/refs"
	"git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/rpc"
	rpcclient "git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/rpc/client"
	"git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/session"
	"git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/signature"
	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/client"
	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/network"
	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/object/internal"
	frostfscrypto "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/crypto"
	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
	oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/trace"
)

type headRequestForwarder struct {
	Request    *objectV2.HeadRequest
	Response   *objectV2.HeadResponse
	OnceResign *sync.Once
	ObjectAddr oid.Address
	Key        *ecdsa.PrivateKey
}

func (f *headRequestForwarder) forwardRequestToNode(ctx context.Context, addr network.Address, c client.MultiAddressClient, pubkey []byte) (*object.Object, error) {
	ctx, span := tracing.StartSpanFromContext(ctx, "headRequestForwarder.forwardRequestToNode",
		trace.WithAttributes(attribute.String("address", addr.String())),
	)
	defer span.End()

	var err error

	// once compose and resign forwarding request
	f.OnceResign.Do(func() {
		// compose meta header of the local server
		metaHdr := new(session.RequestMetaHeader)
		metaHdr.SetTTL(f.Request.GetMetaHeader().GetTTL() - 1)
		// TODO: #1165 think how to set the other fields
		metaHdr.SetOrigin(f.Request.GetMetaHeader())
		writeCurrentVersion(metaHdr)

		f.Request.SetMetaHeader(metaHdr)

		err = signature.SignServiceMessage(f.Key, f.Request)
	})

	if err != nil {
		return nil, err
	}

	headResp, err := f.sendHeadRequest(ctx, addr, c)
	if err != nil {
		return nil, err
	}

	if err := f.verifyResponse(headResp, pubkey); err != nil {
		return nil, err
	}

	var (
		hdr   *objectV2.Header
		idSig *refs.Signature
	)

	switch v := headResp.GetBody().GetHeaderPart().(type) {
	case nil:
		return nil, fmt.Errorf("unexpected header type %T", v)
	case *objectV2.ShortHeader:
		if hdr, err = f.getHeaderFromShortHeader(v); err != nil {
			return nil, err
		}
	case *objectV2.HeaderWithSignature:
		if hdr, idSig, err = f.getHeaderAndSignature(v); err != nil {
			return nil, err
		}
	case *objectV2.SplitInfo:
		si := object.NewSplitInfoFromV2(v)
		return nil, object.NewSplitInfoError(si)
	}

	objv2 := new(objectV2.Object)
	objv2.SetHeader(hdr)
	objv2.SetSignature(idSig)

	obj := object.NewFromV2(objv2)
	obj.SetID(f.ObjectAddr.Object())

	return obj, nil
}

func (f *headRequestForwarder) getHeaderFromShortHeader(sh *objectV2.ShortHeader) (*objectV2.Header, error) {
	if !f.Request.GetBody().GetMainOnly() {
		return nil, fmt.Errorf("wrong header part type: expected %T, received %T",
			(*objectV2.ShortHeader)(nil), (*objectV2.HeaderWithSignature)(nil),
		)
	}

	hdr := new(objectV2.Header)
	hdr.SetPayloadLength(sh.GetPayloadLength())
	hdr.SetVersion(sh.GetVersion())
	hdr.SetOwnerID(sh.GetOwnerID())
	hdr.SetObjectType(sh.GetObjectType())
	hdr.SetCreationEpoch(sh.GetCreationEpoch())
	hdr.SetPayloadHash(sh.GetPayloadHash())
	hdr.SetHomomorphicHash(sh.GetHomomorphicHash())
	return hdr, nil
}

func (f *headRequestForwarder) getHeaderAndSignature(hdrWithSig *objectV2.HeaderWithSignature) (*objectV2.Header, *refs.Signature, error) {
	if f.Request.GetBody().GetMainOnly() {
		return nil, nil, fmt.Errorf("wrong header part type: expected %T, received %T",
			(*objectV2.HeaderWithSignature)(nil), (*objectV2.ShortHeader)(nil),
		)
	}

	if hdrWithSig == nil {
		return nil, nil, errors.New("nil object part")
	}

	hdr := hdrWithSig.GetHeader()
	idSig := hdrWithSig.GetSignature()

	if idSig == nil {
		// TODO(@cthulhu-rider): #1387 use "const" error
		return nil, nil, errors.New("missing signature")
	}

	binID, err := f.ObjectAddr.Object().Marshal()
	if err != nil {
		return nil, nil, fmt.Errorf("marshal ID: %w", err)
	}

	var sig frostfscrypto.Signature
	if err := sig.ReadFromV2(*idSig); err != nil {
		return nil, nil, fmt.Errorf("can't read signature: %w", err)
	}

	if !sig.Verify(binID) {
		return nil, nil, errors.New("invalid object ID signature")
	}

	return hdr, idSig, nil
}

func (f *headRequestForwarder) sendHeadRequest(ctx context.Context, addr network.Address, c client.MultiAddressClient) (*objectV2.HeadResponse, error) {
	var headResp *objectV2.HeadResponse
	err := c.RawForAddress(ctx, addr, func(cli *rpcclient.Client) error {
		var e error
		headResp, e = rpc.HeadObject(cli, f.Request, rpcclient.WithContext(ctx))
		return e
	})
	if err != nil {
		return nil, fmt.Errorf("sending the request failed: %w", err)
	}
	return headResp, nil
}

func (f *headRequestForwarder) verifyResponse(headResp *objectV2.HeadResponse, pubkey []byte) error {
	// verify response key
	if err := internal.VerifyResponseKeyV2(pubkey, headResp); err != nil {
		return err
	}

	// verify response structure
	if err := signature.VerifyServiceMessage(headResp); err != nil {
		return fmt.Errorf("response verification failed: %w", err)
	}

	return checkStatus(f.Response.GetMetaHeader().GetStatus())
}