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

174 lines
4.9 KiB
Go

package getsvc
import (
"context"
"crypto/ecdsa"
"sync"
objectV2 "git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/object"
"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"
"git.frostfs.info/TrueCloudLab/frostfs-observability/tracing"
frostfscrypto "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/crypto"
objectSDK "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) (*objectSDK.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, errUnexpectedHeaderPart(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 := objectSDK.NewSplitInfoFromV2(v)
return nil, objectSDK.NewSplitInfoError(si)
}
objv2 := new(objectV2.Object)
objv2.SetHeader(hdr)
objv2.SetSignature(idSig)
obj := objectSDK.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, errWrongHeaderPartTypeExpShortRecvWithSignature
}
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, errWrongHeaderPartTypeExpWithSignRecvShort
}
if hdrWithSig == nil {
return nil, nil, errNilObjectPart
}
hdr := hdrWithSig.GetHeader()
idSig := hdrWithSig.GetSignature()
if idSig == nil {
return nil, nil, errMissingSignature
}
binID, err := f.ObjectAddr.Object().Marshal()
if err != nil {
return nil, nil, errMarshalID(err)
}
var sig frostfscrypto.Signature
if err := sig.ReadFromV2(*idSig); err != nil {
return nil, nil, errCantReadSignature(err)
}
if !sig.Verify(binID) {
return nil, nil, errInvalidObjectIDSign
}
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, errSendingRequestFailed(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 errResponseVerificationFailed(err)
}
return checkStatus(f.Response.GetMetaHeader().GetStatus())
}