services/object: Inline signing service

Continues 1e40321. Now sign service is gone.

Signed-off-by: Leonard Lyubich <[email protected]>

cmd/neofs-node/object.go

@@ -37,7 +37,6 @@ import (
 	truststorage "github.com/nspcc-dev/neofs-node/pkg/services/reputation/local/storage"
 	"github.com/nspcc-dev/neofs-sdk-go/client"
 	cid "github.com/nspcc-dev/neofs-sdk-go/container/id"
-	neofsecdsa "github.com/nspcc-dev/neofs-sdk-go/crypto/ecdsa"
 	eaclSDK "github.com/nspcc-dev/neofs-sdk-go/eacl"
 	netmapsdk "github.com/nspcc-dev/neofs-sdk-go/netmap"
 	objectSDK "github.com/nspcc-dev/neofs-sdk-go/object"
@@ -294,7 +293,7 @@ func initObjectService(c *cfg) {
 	)
 
 	// build service pipeline
-	// grpc | signature | response | acl | split
+	// grpc | response | acl | split
 
 	splitSvc := objectService.NewTransportSplitter(
 		c.cfgGRPC.maxChunkSize,
@@ -342,12 +341,7 @@ func initObjectService(c *cfg) {
 		c.respSvc,
 	)
 
-	signSvc := objectService.NewSignService(
-		&c.key.PrivateKey,
-		respSvc,
-	)
-
-	server := objectService.New(signSvc, mNumber, objNode, neofsecdsa.SignerRFC6979(c.shared.basics.key.PrivateKey), c.cfgNetmap.state, c.metricsCollector)
+	server := objectService.New(respSvc, mNumber, objNode, c.shared.basics.key.PrivateKey, c.cfgNetmap.state, c.metricsCollector)
 
 	for _, srv := range c.cfgGRPC.servers {
 		objectGRPC.RegisterObjectServiceServer(srv, server)

pkg/services/object/server.go

@@ -3,6 +3,7 @@ package object
 import (
 	"bytes"
 	"context"
+	"crypto/ecdsa"
 	"encoding/binary"
 	"errors"
 	"fmt"
@@ -13,6 +14,8 @@ import (
 	protoobject "github.com/nspcc-dev/neofs-api-go/v2/object/grpc"
 	refsv2 "github.com/nspcc-dev/neofs-api-go/v2/refs"
 	refs "github.com/nspcc-dev/neofs-api-go/v2/refs/grpc"
+	protosession "github.com/nspcc-dev/neofs-api-go/v2/session/grpc"
+	"github.com/nspcc-dev/neofs-api-go/v2/signature"
 	protostatus "github.com/nspcc-dev/neofs-api-go/v2/status/grpc"
 	"github.com/nspcc-dev/neofs-node/pkg/core/netmap"
 	objectcore "github.com/nspcc-dev/neofs-node/pkg/core/object"
@@ -110,14 +113,14 @@ type server struct {
 	srv ServiceServer
 
 	node    Node
-	signer  neofscrypto.Signer
+	signer  ecdsa.PrivateKey
 	mNumber uint32
 	nmState netmap.StateDetailed
 	metrics MetricCollector
 }
 
 // New provides protoobject.ObjectServiceServer for the given parameters.
-func New(c ServiceServer, magicNumber uint32, node Node, signer neofscrypto.Signer, nmState netmap.StateDetailed, m MetricCollector) protoobject.ObjectServiceServer {
+func New(c ServiceServer, magicNumber uint32, node Node, signer ecdsa.PrivateKey, nmState netmap.StateDetailed, m MetricCollector) protoobject.ObjectServiceServer {
 	return &server{
 		srv:     c,
 		node:    node,
@@ -132,6 +135,21 @@ func (s *server) pushOpExecResult(op stat.Method, err error, startedAt time.Time
 	s.metrics.HandleOpExecResult(op, err == nil, time.Since(startedAt))
 }
 
+func (s *server) makeResponseMetaHeader(st *protostatus.Status) *protosession.ResponseMetaHeader {
+	return &protosession.ResponseMetaHeader{Status: st}
+}
+
+func (s *server) sendPutResponse(stream protoobject.ObjectService_PutServer, resp *protoobject.PutResponse) error {
+	resp = util.SignResponse(&s.signer, resp, v2object.PutResponse{})
+	return stream.SendAndClose(resp)
+}
+
+func (s *server) sendStatusPutResponse(stream protoobject.ObjectService_PutServer, err error) error {
+	return s.sendPutResponse(stream, &protoobject.PutResponse{
+		MetaHeader: s.makeResponseMetaHeader(util.ToStatus(err)),
+	})
+}
+
 // Put opens internal Object service Put stream and overtakes data from gRPC stream to it.
 func (s *server) Put(gStream protoobject.ObjectService_PutServer) error {
 	stream, err := s.srv.Put(gStream.Context())
@@ -148,7 +166,9 @@ func (s *server) Put(gStream protoobject.ObjectService_PutServer) error {
 		if req, err = gStream.Recv(); err != nil {
 			if errors.Is(err, io.EOF) {
 				if resp, err = stream.CloseAndRecv(); err == nil {
-					err = gStream.SendAndClose(resp.ToGRPCMessage().(*protoobject.PutResponse)) // assign for defer
+					err = s.sendPutResponse(gStream, resp.ToGRPCMessage().(*protoobject.PutResponse)) // assign for defer
+				} else {
+					err = s.sendStatusPutResponse(gStream, err)
 				}
 			}
 			return err
@@ -162,18 +182,34 @@ func (s *server) Put(gStream protoobject.ObjectService_PutServer) error {
 		if err = putReq.FromGRPCMessage(req); err != nil {
 			return err
 		}
+		if err = signature.VerifyServiceMessage(putReq); err != nil {
+			err = s.sendStatusPutResponse(gStream, util.ToRequestSignatureVerificationError(err)) // assign for defer
+			return err
+		}
 
 		if err = stream.Send(putReq); err != nil {
-			if errors.Is(err, util.ErrAbortStream) {
-				if resp, err = stream.CloseAndRecv(); err == nil {
-					err = gStream.SendAndClose(resp.ToGRPCMessage().(*protoobject.PutResponse)) // assign for defer
-				}
+			if resp, err = stream.CloseAndRecv(); err == nil {
+				err = s.sendPutResponse(gStream, resp.ToGRPCMessage().(*protoobject.PutResponse)) // assign for defer
+			} else {
+				err = s.sendStatusPutResponse(gStream, err)
 			}
 			return err
 		}
 	}
 }
 
+func (s *server) signDeleteResponse(startTime time.Time, err error, resp *protoobject.DeleteResponse) (*protoobject.DeleteResponse, error) {
+	resp = util.SignResponse(&s.signer, resp, v2object.DeleteResponse{})
+	s.pushOpExecResult(stat.MethodObjectDelete, err, startTime)
+	return resp, nil
+}
+
+func (s *server) makeStatusDeleteResponse(startTime time.Time, err error) (*protoobject.DeleteResponse, error) {
+	return s.signDeleteResponse(startTime, err, &protoobject.DeleteResponse{
+		MetaHeader: s.makeResponseMetaHeader(util.ToStatus(err)),
+	})
+}
+
 // Delete converts gRPC DeleteRequest message and passes it to internal Object service.
 func (s *server) Delete(ctx context.Context, req *protoobject.DeleteRequest) (*protoobject.DeleteResponse, error) {
 	delReq := new(v2object.DeleteRequest)
@@ -182,13 +218,27 @@ func (s *server) Delete(ctx context.Context, req *protoobject.DeleteRequest) (*p
 	}
 
 	t := time.Now()
+	if err := signature.VerifyServiceMessage(delReq); err != nil {
+		return s.makeStatusDeleteResponse(t, err)
+	}
 	resp, err := s.srv.Delete(ctx, delReq)
-	s.pushOpExecResult(stat.MethodObjectDelete, err, t)
 	if err != nil {
-		return nil, err
+		return s.makeStatusDeleteResponse(t, err)
 	}
 
-	return resp.ToGRPCMessage().(*protoobject.DeleteResponse), nil
+	return s.signDeleteResponse(t, util.StatusOKErr, resp.ToGRPCMessage().(*protoobject.DeleteResponse))
+}
+
+func (s *server) signHeadResponse(err error, startTime time.Time, resp *protoobject.HeadResponse) (*protoobject.HeadResponse, error) {
+	resp = util.SignResponse(&s.signer, resp, v2object.HeadResponse{})
+	s.pushOpExecResult(stat.MethodObjectHead, err, startTime)
+	return resp, nil
+}
+
+func (s *server) makeStatusHeadResponse(startTime time.Time, err error) (*protoobject.HeadResponse, error) {
+	return s.signHeadResponse(err, startTime, &protoobject.HeadResponse{
+		MetaHeader: s.makeResponseMetaHeader(util.ToStatus(err)),
+	})
 }
 
 // Head converts gRPC HeadRequest message and passes it to internal Object service.
@@ -199,13 +249,27 @@ func (s *server) Head(ctx context.Context, req *protoobject.HeadRequest) (*proto
 	}
 
 	t := time.Now()
+	if err := signature.VerifyServiceMessage(searchReq); err != nil {
+		return s.makeStatusHeadResponse(t, err)
+	}
 	resp, err := s.srv.Head(ctx, searchReq)
-	s.pushOpExecResult(stat.MethodObjectHead, err, t)
 	if err != nil {
-		return nil, err
+		return s.makeStatusHeadResponse(t, err)
 	}
 
-	return resp.ToGRPCMessage().(*protoobject.HeadResponse), nil
+	return s.signHeadResponse(nil, t, resp.ToGRPCMessage().(*protoobject.HeadResponse))
+}
+
+func (s *server) signHashResponse(startTime time.Time, err error, resp *protoobject.GetRangeHashResponse) (*protoobject.GetRangeHashResponse, error) {
+	resp = util.SignResponse(&s.signer, resp, v2object.GetRangeHashResponse{})
+	s.pushOpExecResult(stat.MethodObjectHash, err, startTime)
+	return resp, nil
+}
+
+func (s *server) makeStatusHashResponse(startTime time.Time, err error) (*protoobject.GetRangeHashResponse, error) {
+	return s.signHashResponse(startTime, err, &protoobject.GetRangeHashResponse{
+		MetaHeader: s.makeResponseMetaHeader(util.ToStatus(err)),
+	})
 }
 
 // GetRangeHash converts gRPC GetRangeHashRequest message and passes it to internal Object service.
@@ -216,26 +280,42 @@ func (s *server) GetRangeHash(ctx context.Context, req *protoobject.GetRangeHash
 	}
 
 	t := time.Now()
+	if err := signature.VerifyServiceMessage(hashRngReq); err != nil {
+		return s.makeStatusHashResponse(t, err)
+	}
 	resp, err := s.srv.GetRangeHash(ctx, hashRngReq)
 	s.pushOpExecResult(stat.MethodObjectHash, err, t)
 	if err != nil {
-		return nil, err
+		return s.makeStatusHashResponse(t, err)
 	}
 
-	return resp.ToGRPCMessage().(*protoobject.GetRangeHashResponse), nil
+	return s.signHashResponse(t, nil, resp.ToGRPCMessage().(*protoobject.GetRangeHashResponse))
+}
+
+func (s *server) sendGetResponse(stream protoobject.ObjectService_GetServer, resp *protoobject.GetResponse) error {
+	resp = util.SignResponse(&s.signer, resp, v2object.GetResponse{})
+	return stream.Send(resp)
+}
+
+func (s *server) sendStatusGetResponse(startTime time.Time, stream protoobject.ObjectService_GetServer, err error) error {
+	sendErr := s.sendGetResponse(stream, &protoobject.GetResponse{
+		MetaHeader: s.makeResponseMetaHeader(util.ToStatus(err)),
+	})
+	s.pushOpExecResult(stat.MethodObjectGet, err, startTime)
+	return sendErr
 }
 
 type getStreamerV2 struct {
 	protoobject.ObjectService_GetServer
-	metrics MetricCollector
+	srv *server
 }
 
 func (s *getStreamerV2) Send(resp *v2object.GetResponse) error {
 	r := resp.ToGRPCMessage().(*protoobject.GetResponse)
 	if c := r.GetBody().GetChunk(); c != nil {
-		s.metrics.AddGetPayload(len(c))
+		s.srv.metrics.AddGetPayload(len(c))
 	}
-	return s.ObjectService_GetServer.Send(r)
+	return s.srv.sendGetResponse(s.ObjectService_GetServer, r)
 }
 
 // Get converts gRPC GetRequest message and server-side stream and overtakes its data
@@ -246,25 +326,43 @@ func (s *server) Get(req *protoobject.GetRequest, gStream protoobject.ObjectServ
 		return err
 	}
 	t := time.Now()
+	if err := signature.VerifyServiceMessage(getReq); err != nil {
+		return s.sendStatusGetResponse(t, gStream, err)
+	}
 	err := s.srv.Get(
 		getReq,
 		&getStreamerV2{
 			ObjectService_GetServer: gStream,
-			metrics:                 s.metrics,
+			srv:                     s,
 		},
 	)
-	s.pushOpExecResult(stat.MethodObjectGet, err, t)
-	return err
+	if err != nil {
+		return s.sendStatusGetResponse(t, gStream, err)
+	}
+	s.pushOpExecResult(stat.MethodObjectGet, nil, t)
+	return nil
+}
+
+func (s *server) sendRangeResponse(stream protoobject.ObjectService_GetRangeServer, resp *protoobject.GetRangeResponse) error {
+	resp = util.SignResponse(&s.signer, resp, v2object.GetRangeResponse{})
+	return stream.Send(resp)
+}
+
+func (s *server) sendStatusRangeResponse(startTime time.Time, stream protoobject.ObjectService_GetRangeServer, err error) error {
+	sendErr := s.sendRangeResponse(stream, &protoobject.GetRangeResponse{
+		MetaHeader: s.makeResponseMetaHeader(util.ToStatus(err)),
+	})
+	s.pushOpExecResult(stat.MethodObjectRange, err, startTime)
+	return sendErr
 }
 
 type getRangeStreamerV2 struct {
 	protoobject.ObjectService_GetRangeServer
+	srv *server
 }
 
 func (s *getRangeStreamerV2) Send(resp *v2object.GetRangeResponse) error {
-	return s.ObjectService_GetRangeServer.Send(
-		resp.ToGRPCMessage().(*protoobject.GetRangeResponse),
-	)
+	return s.srv.sendRangeResponse(s.ObjectService_GetRangeServer, resp.ToGRPCMessage().(*protoobject.GetRangeResponse))
 }
 
 // GetRange converts gRPC GetRangeRequest message and server-side stream and overtakes its data
@@ -275,24 +373,43 @@ func (s *server) GetRange(req *protoobject.GetRangeRequest, gStream protoobject.
 		return err
 	}
 	t := time.Now()
+	if err := signature.VerifyServiceMessage(getRngReq); err != nil {
+		return s.sendStatusRangeResponse(t, gStream, err)
+	}
 	err := s.srv.GetRange(
 		getRngReq,
 		&getRangeStreamerV2{
 			ObjectService_GetRangeServer: gStream,
+			srv:                          s,
 		},
 	)
-	s.pushOpExecResult(stat.MethodObjectRange, err, t)
-	return err
+	if err != nil {
+		return s.sendStatusRangeResponse(t, gStream, err)
+	}
+	s.pushOpExecResult(stat.MethodObjectRange, nil, t)
+	return nil
+}
+
+func (s *server) sendSearchResponse(stream protoobject.ObjectService_SearchServer, resp *protoobject.SearchResponse) error {
+	resp = util.SignResponse(&s.signer, resp, v2object.SearchResponse{})
+	return stream.Send(resp)
+}
+
+func (s *server) sendStatusSearchResponse(startTime time.Time, stream protoobject.ObjectService_SearchServer, err error) error {
+	sendErr := s.sendSearchResponse(stream, &protoobject.SearchResponse{
+		MetaHeader: s.makeResponseMetaHeader(util.ToStatus(err)),
+	})
+	s.pushOpExecResult(stat.MethodObjectSearch, err, startTime)
+	return sendErr
 }
 
 type searchStreamerV2 struct {
 	protoobject.ObjectService_SearchServer
+	srv *server
 }
 
 func (s *searchStreamerV2) Send(resp *v2object.SearchResponse) error {
-	return s.ObjectService_SearchServer.Send(
-		resp.ToGRPCMessage().(*protoobject.SearchResponse),
-	)
+	return s.srv.sendSearchResponse(s.ObjectService_SearchServer, resp.ToGRPCMessage().(*protoobject.SearchResponse))
 }
 
 // Search converts gRPC SearchRequest message and server-side stream and overtakes its data
@@ -303,14 +420,21 @@ func (s *server) Search(req *protoobject.SearchRequest, gStream protoobject.Obje
 		return err
 	}
 	t := time.Now()
+	if err := signature.VerifyServiceMessage(searchReq); err != nil {
+		return s.sendStatusSearchResponse(t, gStream, err)
+	}
 	err := s.srv.Search(
 		searchReq,
 		&searchStreamerV2{
 			ObjectService_SearchServer: gStream,
+			srv:                        s,
 		},
 	)
-	s.pushOpExecResult(stat.MethodObjectSearch, err, t)
-	return err
+	if err != nil {
+		return s.sendStatusSearchResponse(t, gStream, err)
+	}
+	s.pushOpExecResult(stat.MethodObjectSearch, nil, t)
+	return nil
 }
 
 // Replicate serves neo.fs.v2.object.ObjectService/Replicate RPC.
@@ -542,16 +666,16 @@ func (s *server) metaInfoSignature(o object.Object) ([]byte, error) {
 	var firstSig neofscrypto.Signature
 	var secondSig neofscrypto.Signature
 	var thirdSig neofscrypto.Signature
-
-	err := firstSig.Calculate(s.signer, firstMeta)
+	signer := neofsecdsa.SignerRFC6979(s.signer)
+	err := firstSig.Calculate(signer, firstMeta)
 	if err != nil {
 		return nil, fmt.Errorf("signature failure: %w", err)
 	}
-	err = secondSig.Calculate(s.signer, secondMeta)
+	err = secondSig.Calculate(signer, secondMeta)
 	if err != nil {
 		return nil, fmt.Errorf("signature failure: %w", err)
 	}
-	err = thirdSig.Calculate(s.signer, thirdMeta)
+	err = thirdSig.Calculate(signer, thirdMeta)
 	if err != nil {
 		return nil, fmt.Errorf("signature failure: %w", err)
 	}