package getsvc import ( "context" "crypto/sha256" "errors" "fmt" "hash" "io" "sync" objectV2 "github.com/nspcc-dev/neofs-api-go/v2/object" "github.com/nspcc-dev/neofs-api-go/v2/refs" "github.com/nspcc-dev/neofs-api-go/v2/rpc" rpcclient "github.com/nspcc-dev/neofs-api-go/v2/rpc/client" "github.com/nspcc-dev/neofs-api-go/v2/session" "github.com/nspcc-dev/neofs-api-go/v2/signature" "github.com/nspcc-dev/neofs-node/pkg/core/client" "github.com/nspcc-dev/neofs-node/pkg/network" objectSvc "github.com/nspcc-dev/neofs-node/pkg/services/object" getsvc "github.com/nspcc-dev/neofs-node/pkg/services/object/get" "github.com/nspcc-dev/neofs-node/pkg/services/object/internal" "github.com/nspcc-dev/neofs-node/pkg/services/object/util" "github.com/nspcc-dev/neofs-sdk-go/object" addressSDK "github.com/nspcc-dev/neofs-sdk-go/object/address" signature2 "github.com/nspcc-dev/neofs-sdk-go/util/signature" "github.com/nspcc-dev/tzhash/tz" ) var errWrongMessageSeq = errors.New("incorrect message sequence") func (s *Service) toPrm(req *objectV2.GetRequest, stream objectSvc.GetObjectStream) (*getsvc.Prm, error) { meta := req.GetMetaHeader() commonPrm, err := util.CommonPrmFromV2(req) if err != nil { return nil, err } p := new(getsvc.Prm) p.SetCommonParameters(commonPrm) body := req.GetBody() p.WithAddress(addressSDK.NewAddressFromV2(body.GetAddress())) p.WithRawFlag(body.GetRaw()) p.SetObjectWriter(&streamObjectWriter{stream}) if !commonPrm.LocalOnly() { var onceResign sync.Once p.SetRequestForwarder(groupAddressRequestForwarder(func(addr network.Address, c client.MultiAddressClient, pubkey []byte) (*object.Object, error) { var err error key, err := s.keyStorage.GetKey(nil) if err != nil { return nil, err } // once compose and resign forwarding request onceResign.Do(func() { // compose meta header of the local server metaHdr := new(session.RequestMetaHeader) metaHdr.SetTTL(meta.GetTTL() - 1) // TODO: #1165 think how to set the other fields metaHdr.SetOrigin(meta) req.SetMetaHeader(metaHdr) err = signature.SignServiceMessage(key, req) }) if err != nil { return nil, err } // code below is copy-pasted from c.GetObject implementation, // perhaps it is worth highlighting the utility function in neofs-api-go // open stream var getStream *rpc.GetResponseReader err = c.RawForAddress(addr, func(cli *rpcclient.Client) error { getStream, err = rpc.GetObject(cli, req, rpcclient.WithContext(stream.Context())) return err }) if err != nil { return nil, fmt.Errorf("stream opening failed: %w", err) } var ( headWas bool payload []byte obj = new(objectV2.Object) resp = new(objectV2.GetResponse) ) for { // receive message from server stream err := getStream.Read(resp) if err != nil { if errors.Is(err, io.EOF) { if !headWas { return nil, io.ErrUnexpectedEOF } break } return nil, fmt.Errorf("reading the response failed: %w", err) } // verify response key if err = internal.VerifyResponseKeyV2(pubkey, resp); err != nil { return nil, err } // verify response structure if err := signature.VerifyServiceMessage(resp); err != nil { return nil, fmt.Errorf("response verification failed: %w", err) } switch v := resp.GetBody().GetObjectPart().(type) { default: return nil, fmt.Errorf("unexpected object part %T", v) case *objectV2.GetObjectPartInit: if headWas { return nil, errWrongMessageSeq } headWas = true obj.SetObjectID(v.GetObjectID()) obj.SetSignature(v.GetSignature()) hdr := v.GetHeader() obj.SetHeader(hdr) payload = make([]byte, 0, hdr.GetPayloadLength()) case *objectV2.GetObjectPartChunk: if !headWas { return nil, errWrongMessageSeq } payload = append(payload, v.GetChunk()...) case *objectV2.SplitInfo: si := object.NewSplitInfoFromV2(v) return nil, object.NewSplitInfoError(si) } } obj.SetPayload(payload) // convert the object return object.NewFromV2(obj), nil })) } return p, nil } func (s *Service) toRangePrm(req *objectV2.GetRangeRequest, stream objectSvc.GetObjectRangeStream) (*getsvc.RangePrm, error) { meta := req.GetMetaHeader() commonPrm, err := util.CommonPrmFromV2(req) if err != nil { return nil, err } p := new(getsvc.RangePrm) p.SetCommonParameters(commonPrm) body := req.GetBody() p.WithAddress(addressSDK.NewAddressFromV2(body.GetAddress())) p.WithRawFlag(body.GetRaw()) p.SetChunkWriter(&streamObjectRangeWriter{stream}) p.SetRange(object.NewRangeFromV2(body.GetRange())) if !commonPrm.LocalOnly() { var onceResign sync.Once key, err := s.keyStorage.GetKey(nil) if err != nil { return nil, err } p.SetRequestForwarder(groupAddressRequestForwarder(func(addr network.Address, c client.MultiAddressClient, pubkey []byte) (*object.Object, error) { var err error // once compose and resign forwarding request onceResign.Do(func() { // compose meta header of the local server metaHdr := new(session.RequestMetaHeader) metaHdr.SetTTL(meta.GetTTL() - 1) // TODO: #1165 think how to set the other fields metaHdr.SetOrigin(meta) req.SetMetaHeader(metaHdr) err = signature.SignServiceMessage(key, req) }) if err != nil { return nil, err } // code below is copy-pasted from c.ObjectPayloadRangeData implementation, // perhaps it is worth highlighting the utility function in neofs-api-go // open stream var rangeStream *rpc.ObjectRangeResponseReader err = c.RawForAddress(addr, func(cli *rpcclient.Client) error { rangeStream, err = rpc.GetObjectRange(cli, req, rpcclient.WithContext(stream.Context())) return err }) if err != nil { return nil, fmt.Errorf("could not create Get payload range stream: %w", err) } payload := make([]byte, 0, body.GetRange().GetLength()) resp := new(objectV2.GetRangeResponse) for { // receive message from server stream err := rangeStream.Read(resp) if err != nil { if errors.Is(err, io.EOF) { break } return nil, fmt.Errorf("reading the response failed: %w", err) } // verify response key if err = internal.VerifyResponseKeyV2(pubkey, resp); err != nil { return nil, err } // verify response structure if err := signature.VerifyServiceMessage(resp); err != nil { return nil, fmt.Errorf("could not verify %T: %w", resp, err) } switch v := resp.GetBody().GetRangePart().(type) { case nil: return nil, fmt.Errorf("unexpected range type %T", v) case *objectV2.GetRangePartChunk: payload = append(payload, v.GetChunk()...) case *objectV2.SplitInfo: si := object.NewSplitInfoFromV2(v) return nil, object.NewSplitInfoError(si) } } obj := object.New() obj.SetPayload(payload) return obj, nil })) } return p, nil } func (s *Service) toHashRangePrm(req *objectV2.GetRangeHashRequest) (*getsvc.RangeHashPrm, error) { commonPrm, err := util.CommonPrmFromV2(req) if err != nil { return nil, err } p := new(getsvc.RangeHashPrm) p.SetCommonParameters(commonPrm) body := req.GetBody() p.WithAddress(addressSDK.NewAddressFromV2(body.GetAddress())) rngsV2 := body.GetRanges() rngs := make([]object.Range, len(rngsV2)) for i := range rngsV2 { rngs[i] = *object.NewRangeFromV2(&rngsV2[i]) } p.SetRangeList(rngs) p.SetSalt(body.GetSalt()) switch t := body.GetType(); t { default: return nil, fmt.Errorf("unknown checksum type %v", t) case refs.SHA256: p.SetHashGenerator(func() hash.Hash { return sha256.New() }) case refs.TillichZemor: p.SetHashGenerator(func() hash.Hash { return tz.New() }) } return p, nil } type headResponseWriter struct { mainOnly bool body *objectV2.HeadResponseBody } func (w *headResponseWriter) WriteHeader(hdr *object.Object) error { if w.mainOnly { w.body.SetHeaderPart(toShortObjectHeader(hdr)) } else { w.body.SetHeaderPart(toFullObjectHeader(hdr)) } return nil } func (s *Service) toHeadPrm(ctx context.Context, req *objectV2.HeadRequest, resp *objectV2.HeadResponse) (*getsvc.HeadPrm, error) { meta := req.GetMetaHeader() commonPrm, err := util.CommonPrmFromV2(req) if err != nil { return nil, err } p := new(getsvc.HeadPrm) p.SetCommonParameters(commonPrm) body := req.GetBody() objAddr := addressSDK.NewAddressFromV2(body.GetAddress()) p.WithAddress(objAddr) p.WithRawFlag(body.GetRaw()) p.SetHeaderWriter(&headResponseWriter{ mainOnly: body.GetMainOnly(), body: resp.GetBody(), }) if !commonPrm.LocalOnly() { var onceResign sync.Once p.SetRequestForwarder(groupAddressRequestForwarder(func(addr network.Address, c client.MultiAddressClient, pubkey []byte) (*object.Object, error) { var err error key, err := s.keyStorage.GetKey(nil) if err != nil { return nil, err } // once compose and resign forwarding request onceResign.Do(func() { // compose meta header of the local server metaHdr := new(session.RequestMetaHeader) metaHdr.SetTTL(meta.GetTTL() - 1) // TODO: #1165 think how to set the other fields metaHdr.SetOrigin(meta) req.SetMetaHeader(metaHdr) err = signature.SignServiceMessage(key, req) }) if err != nil { return nil, err } // code below is copy-pasted from c.GetObjectHeader implementation, // perhaps it is worth highlighting the utility function in neofs-api-go // send Head request var headResp *objectV2.HeadResponse err = c.RawForAddress(addr, func(cli *rpcclient.Client) error { headResp, err = rpc.HeadObject(cli, req, rpcclient.WithContext(ctx)) return err }) if err != nil { return nil, fmt.Errorf("sending the request failed: %w", err) } // verify response key if err = internal.VerifyResponseKeyV2(pubkey, headResp); err != nil { return nil, err } // verify response structure if err := signature.VerifyServiceMessage(headResp); err != nil { return nil, fmt.Errorf("response verification failed: %w", 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 !body.GetMainOnly() { return nil, fmt.Errorf("wrong header part type: expected %T, received %T", (*objectV2.ShortHeader)(nil), (*objectV2.HeaderWithSignature)(nil), ) } h := v hdr = new(objectV2.Header) hdr.SetPayloadLength(h.GetPayloadLength()) hdr.SetVersion(h.GetVersion()) hdr.SetOwnerID(h.GetOwnerID()) hdr.SetObjectType(h.GetObjectType()) hdr.SetCreationEpoch(h.GetCreationEpoch()) hdr.SetPayloadHash(h.GetPayloadHash()) hdr.SetHomomorphicHash(h.GetHomomorphicHash()) case *objectV2.HeaderWithSignature: if body.GetMainOnly() { return nil, fmt.Errorf("wrong header part type: expected %T, received %T", (*objectV2.HeaderWithSignature)(nil), (*objectV2.ShortHeader)(nil), ) } hdrWithSig := v if hdrWithSig == nil { return nil, errors.New("nil object part") } hdr = hdrWithSig.GetHeader() idSig = hdrWithSig.GetSignature() id, ok := objAddr.ObjectID() if !ok { return nil, errors.New("missing object ID") } var idV2 refs.ObjectID id.WriteToV2(&idV2) if err := signature2.VerifyDataWithSource( signature.StableMarshalerWrapper{ SM: &idV2, }, func() (key, sig []byte) { return idSig.GetKey(), idSig.GetSign() }, ); err != nil { return nil, fmt.Errorf("incorrect object header signature: %w", err) } case *objectV2.SplitInfo: si := object.NewSplitInfoFromV2(v) return nil, object.NewSplitInfoError(si) } objv2 := new(objectV2.Object) objv2.SetHeader(hdr) objv2.SetSignature(idSig) id, _ := objAddr.ObjectID() obj := object.NewFromV2(objv2) obj.SetID(id) // convert the object return obj, nil })) } return p, nil } func splitInfoResponse(info *object.SplitInfo) *objectV2.GetResponse { resp := new(objectV2.GetResponse) body := new(objectV2.GetResponseBody) resp.SetBody(body) body.SetObjectPart(info.ToV2()) return resp } func splitInfoRangeResponse(info *object.SplitInfo) *objectV2.GetRangeResponse { resp := new(objectV2.GetRangeResponse) body := new(objectV2.GetRangeResponseBody) resp.SetBody(body) body.SetRangePart(info.ToV2()) return resp } func setSplitInfoHeadResponse(info *object.SplitInfo, resp *objectV2.HeadResponse) { resp.GetBody().SetHeaderPart(info.ToV2()) } func toHashResponse(typ refs.ChecksumType, res *getsvc.RangeHashRes) *objectV2.GetRangeHashResponse { resp := new(objectV2.GetRangeHashResponse) body := new(objectV2.GetRangeHashResponseBody) resp.SetBody(body) body.SetType(typ) body.SetHashList(res.Hashes()) return resp } func toFullObjectHeader(hdr *object.Object) objectV2.GetHeaderPart { obj := hdr.ToV2() hs := new(objectV2.HeaderWithSignature) hs.SetHeader(obj.GetHeader()) hs.SetSignature(obj.GetSignature()) return hs } func toShortObjectHeader(hdr *object.Object) objectV2.GetHeaderPart { hdrV2 := hdr.ToV2().GetHeader() sh := new(objectV2.ShortHeader) sh.SetOwnerID(hdrV2.GetOwnerID()) sh.SetCreationEpoch(hdrV2.GetCreationEpoch()) sh.SetPayloadLength(hdrV2.GetPayloadLength()) sh.SetVersion(hdrV2.GetVersion()) sh.SetObjectType(hdrV2.GetObjectType()) sh.SetHomomorphicHash(hdrV2.GetHomomorphicHash()) sh.SetPayloadHash(hdrV2.GetPayloadHash()) return sh } func groupAddressRequestForwarder(f func(network.Address, client.MultiAddressClient, []byte) (*object.Object, error)) getsvc.RequestForwarder { return func(info client.NodeInfo, c client.MultiAddressClient) (*object.Object, error) { var ( firstErr error res *object.Object key = info.PublicKey() ) info.AddressGroup().IterateAddresses(func(addr network.Address) (stop bool) { var err error defer func() { stop = err == nil if stop || firstErr == nil { firstErr = err } // would be nice to log otherwise }() res, err = f(addr, c, key) return }) return res, firstErr } }