storagenode: Report gRPC error when satellite is untrusted (#2658)

* storagenode/piecestore: Unexport endpoint method
  Make an exported endpoint method to be unexported because it's only used
  by the same package and makes easy to change without thinking in
  breaking changes.
* uplink/ecclient: Use structured logger
  Swap sugared logger by the normal structured logger for having the full
  stack traces of the error in the debug message.
* storagenode/piecestore: Send gRPC error codes upload
  Refactoring in the storagenode/piecestore to send gRPC status error codes
  when some of the methods involved by upload return an error.
  
  The uplink related to uploads has also been modified to retrieve the
  gRPC status code when an error is returned by the server.

storagenode/piecestore/endpoint.go

@@ -158,7 +158,7 @@ func (endpoint *Endpoint) Delete(ctx context.Context, delete *pb.PieceDeleteRequ
 		return nil, Error.New("expected delete action got %v", delete.Limit.Action) // TODO: report grpc status unauthorized or bad request
 	}
 
-	if err := endpoint.VerifyOrderLimit(ctx, delete.Limit); err != nil {
+	if err := endpoint.verifyOrderLimit(ctx, delete.Limit); err != nil {
 		// TODO: report grpc status unauthorized or bad request
 		return nil, Error.Wrap(err)
 	}
@@ -218,8 +218,8 @@ func (endpoint *Endpoint) Upload(stream pb.Piecestore_UploadServer) (err error)
 		return ErrProtocol.New("expected put or put repair action got %v", limit.Action) // TODO: report grpc status unauthorized or bad request
 	}
 
-	if err := endpoint.VerifyOrderLimit(ctx, limit); err != nil {
+	if err := endpoint.verifyOrderLimit(ctx, limit); err != nil {
-		return err // TODO: report grpc status unauthorized or bad request
+		return err
 	}
 
 	var pieceWriter *pieces.Writer
@@ -410,7 +410,7 @@ func (endpoint *Endpoint) Download(stream pb.Piecestore_DownloadServer) (err err
 		return ErrProtocol.New("requested more that order limit allows, limit=%v requested=%v", limit.Limit, chunk.ChunkSize)
 	}
 
-	if err := endpoint.VerifyOrderLimit(ctx, limit); err != nil {
+	if err := endpoint.verifyOrderLimit(ctx, limit); err != nil {
 		return Error.Wrap(err) // TODO: report grpc status unauthorized or bad request
 	}
 

storagenode/piecestore/verification.go

@@ -9,6 +9,8 @@ import (
 	"time"
 
 	"github.com/zeebo/errs"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 
 	"storj.io/storj/internal/errs2"
 	"storj.io/storj/pkg/pb"
@@ -26,41 +28,42 @@ var (
 
 // VerifyOrderLimit verifies that the order limit is properly signed and has sane values.
 // It also verifies that the serial number has not been used.
-func (endpoint *Endpoint) VerifyOrderLimit(ctx context.Context, limit *pb.OrderLimit) (err error) {
+func (endpoint *Endpoint) verifyOrderLimit(ctx context.Context, limit *pb.OrderLimit) (err error) {
 	defer mon.Task()(&ctx)(&err)
 
 	// sanity checks
 	now := time.Now()
 	switch {
 	case limit.Limit < 0:
-		return ErrProtocol.New("order limit is negative")
+		return status.Error(codes.InvalidArgument, "order limit is negative")
 	case endpoint.signer.ID() != limit.StorageNodeId:
-		return ErrProtocol.New("order intended for other storagenode: %v", limit.StorageNodeId)
+		return status.Errorf(codes.InvalidArgument, "order intended for other storagenode: %v", limit.StorageNodeId)
 	case endpoint.IsExpired(limit.PieceExpiration):
-		return ErrProtocol.New("piece expired: %v", limit.PieceExpiration)
+		return status.Errorf(codes.InvalidArgument, "piece expired: %v", limit.PieceExpiration)
 	case endpoint.IsExpired(limit.OrderExpiration):
-		return ErrProtocol.New("order expired: %v", limit.OrderExpiration)
+		return status.Errorf(codes.InvalidArgument, "order expired: %v", limit.OrderExpiration)
 	case now.Sub(limit.OrderCreation) > endpoint.config.OrderLimitGracePeriod:
-		return ErrProtocol.New("order created too long ago: %v", limit.OrderCreation)
+		return status.Errorf(codes.InvalidArgument, "order created too long ago: %v", limit.OrderCreation)
 	case limit.SatelliteId.IsZero():
-		return ErrProtocol.New("missing satellite id")
+		return status.Errorf(codes.InvalidArgument, "missing satellite id")
 	case limit.UplinkPublicKey.IsZero():
-		return ErrProtocol.New("missing uplink public key")
+		return status.Errorf(codes.InvalidArgument, "missing uplink public key")
 	case len(limit.SatelliteSignature) == 0:
-		return ErrProtocol.New("missing satellite signature")
+		return status.Errorf(codes.InvalidArgument, "missing satellite signature")
 	case limit.PieceId.IsZero():
-		return ErrProtocol.New("missing piece id")
+		return status.Errorf(codes.InvalidArgument, "missing piece id")
 	}
 
 	if err := endpoint.trust.VerifySatelliteID(ctx, limit.SatelliteId); err != nil {
-		return ErrVerifyUntrusted.Wrap(err)
+		return status.Errorf(codes.PermissionDenied, "untrusted: %+v", err)
 	}
 
 	if err := endpoint.VerifyOrderLimitSignature(ctx, limit); err != nil {
-		if err == context.Canceled {
+		if errs2.IsCanceled(err) {
-			return err
+			return status.Error(codes.Canceled, "context has been canceled")
 		}
-		return ErrVerifyUntrusted.Wrap(err)
+
+		return status.Errorf(codes.Unauthenticated, "untrusted: %+v", err)
 	}
 
 	serialExpiration := limit.OrderExpiration
@@ -71,7 +74,7 @@ func (endpoint *Endpoint) VerifyOrderLimit(ctx context.Context, limit *pb.OrderL
 	}
 
 	if err := endpoint.usedSerials.Add(ctx, limit.SatelliteId, limit.SerialNumber, serialExpiration); err != nil {
-		return ErrVerifyDuplicateRequest.Wrap(err)
+		return status.Errorf(codes.Unauthenticated, "serial number is already used: %+v", err)
 	}
 
 	return nil

uplink/ecclient/client.go

@@ -271,14 +271,22 @@ func (ec *ecClient) putPiece(ctx, parent context.Context, limit *pb.AddressedOrd
 		Address: limit.GetStorageNodeAddress(),
 	})
 	if err != nil {
-		ec.log.Sugar().Debugf("Failed dialing for putting piece %s to node %s: %v", pieceID, storageNodeID, err)
+		ec.log.Debug("Failed dialing for putting piece to node",
+			zap.String("pieceID", pieceID.String()),
+			zap.String("nodeID", storageNodeID.String()),
+			zap.Error(err),
+		)
 		return nil, err
 	}
 	defer func() { err = errs.Combine(err, ps.Close()) }()
 
 	upload, err := ps.Upload(ctx, limit.GetLimit(), privateKey)
 	if err != nil {
-		ec.log.Sugar().Debugf("Failed requesting upload of piece %s to node %s: %v", pieceID, storageNodeID, err)
+		ec.log.Debug("Failed requesting upload of pieces to node",
+			zap.String("pieceID", pieceID.String()),
+			zap.String("nodeID", storageNodeID.String()),
+			zap.Error(err),
+		)
 		return nil, err
 	}
 	defer func() {
@@ -307,7 +315,13 @@ func (ec *ecClient) putPiece(ctx, parent context.Context, limit *pb.AddressedOrd
 		if limit.GetStorageNodeAddress() != nil {
 			nodeAddress = limit.GetStorageNodeAddress().GetAddress()
 		}
-		ec.log.Sugar().Debugf("Failed uploading piece %s to node %s (%+v): %v", pieceID, storageNodeID, nodeAddress, err)
+
+		ec.log.Debug("Failed uploading piece to node",
+			zap.String("pieceID", pieceID.String()),
+			zap.String("nodeID", storageNodeID.String()),
+			zap.String("nodeAddress", nodeAddress),
+			zap.Error(err),
+		)
 	}
 
 	return hash, err

uplink/piecestore/upload.go

@@ -67,7 +67,14 @@ func (client *Client) Upload(ctx context.Context, limit *pb.OrderLimit, piecePri
 	})
 	if err != nil {
 		_, closeErr := stream.CloseAndRecv()
-		return nil, ErrProtocol.Wrap(errs.Combine(err, closeErr))
+		switch {
+		case err != io.EOF && closeErr != nil:
+			err = ErrProtocol.Wrap(errs.Combine(err, closeErr))
+		case closeErr != nil:
+			err = ErrProtocol.Wrap(closeErr)
+		}
+
+		return nil, err
 	}
 
 	upload := &Upload{
@@ -138,7 +145,14 @@ func (client *Upload) Write(data []byte) (written int, err error) {
 			},
 		})
 		if err != nil {
-			err = ErrProtocol.Wrap(err)
+			_, closeErr := client.stream.CloseAndRecv()
+			switch {
+			case err != io.EOF && closeErr != nil:
+				err = ErrProtocol.Wrap(errs.Combine(err, closeErr))
+			case closeErr != nil:
+				err = ErrProtocol.Wrap(closeErr)
+			}
+
 			client.sendError = err
 			return written, err
 		}