Merge pull request #31 from anytypeio/coordinator

Coordinator

Makefile

@@ -17,6 +17,7 @@ proto:
 	protoc --gogofaster_out=$(PKGMAP):. --go-drpc_out=protolib=github.com/gogo/protobuf:. commonfile/fileproto/protos/*.proto
 	protoc --gogofaster_out=$(PKGMAP):. --go-drpc_out=protolib=github.com/gogo/protobuf:. net/streampool/testservice/protos/*.proto
 	protoc --gogofaster_out=:. net/secureservice/handshake/handshakeproto/protos/*.proto
+	protoc --gogofaster_out=$(PKGMAP):. --go-drpc_out=protolib=github.com/gogo/protobuf:. coordinator/coordinatorproto/protos/*.proto
 
 deps:
 	go mod download

coordinator/coordinatorproto/coordinator_drpc.pb.go

@@ -0,0 +1,153 @@
+// Code generated by protoc-gen-go-drpc. DO NOT EDIT.
+// protoc-gen-go-drpc version: v0.0.32
+// source: coordinator/coordinatorproto/protos/coordinator.proto
+
+package coordinatorproto
+
+import (
+	bytes "bytes"
+	context "context"
+	errors "errors"
+	jsonpb "github.com/gogo/protobuf/jsonpb"
+	proto "github.com/gogo/protobuf/proto"
+	drpc "storj.io/drpc"
+	drpcerr "storj.io/drpc/drpcerr"
+)
+
+type drpcEncoding_File_coordinator_coordinatorproto_protos_coordinator_proto struct{}
+
+func (drpcEncoding_File_coordinator_coordinatorproto_protos_coordinator_proto) Marshal(msg drpc.Message) ([]byte, error) {
+	return proto.Marshal(msg.(proto.Message))
+}
+
+func (drpcEncoding_File_coordinator_coordinatorproto_protos_coordinator_proto) Unmarshal(buf []byte, msg drpc.Message) error {
+	return proto.Unmarshal(buf, msg.(proto.Message))
+}
+
+func (drpcEncoding_File_coordinator_coordinatorproto_protos_coordinator_proto) JSONMarshal(msg drpc.Message) ([]byte, error) {
+	var buf bytes.Buffer
+	err := new(jsonpb.Marshaler).Marshal(&buf, msg.(proto.Message))
+	if err != nil {
+		return nil, err
+	}
+	return buf.Bytes(), nil
+}
+
+func (drpcEncoding_File_coordinator_coordinatorproto_protos_coordinator_proto) JSONUnmarshal(buf []byte, msg drpc.Message) error {
+	return jsonpb.Unmarshal(bytes.NewReader(buf), msg.(proto.Message))
+}
+
+type DRPCCoordinatorClient interface {
+	DRPCConn() drpc.Conn
+
+	SpaceSign(ctx context.Context, in *SpaceSignRequest) (*SpaceSignResponse, error)
+	FileLimitCheck(ctx context.Context, in *FileLimitCheckRequest) (*FileLimitCheckResponse, error)
+}
+
+type drpcCoordinatorClient struct {
+	cc drpc.Conn
+}
+
+func NewDRPCCoordinatorClient(cc drpc.Conn) DRPCCoordinatorClient {
+	return &drpcCoordinatorClient{cc}
+}
+
+func (c *drpcCoordinatorClient) DRPCConn() drpc.Conn { return c.cc }
+
+func (c *drpcCoordinatorClient) SpaceSign(ctx context.Context, in *SpaceSignRequest) (*SpaceSignResponse, error) {
+	out := new(SpaceSignResponse)
+	err := c.cc.Invoke(ctx, "/coordinator.Coordinator/SpaceSign", drpcEncoding_File_coordinator_coordinatorproto_protos_coordinator_proto{}, in, out)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *drpcCoordinatorClient) FileLimitCheck(ctx context.Context, in *FileLimitCheckRequest) (*FileLimitCheckResponse, error) {
+	out := new(FileLimitCheckResponse)
+	err := c.cc.Invoke(ctx, "/coordinator.Coordinator/FileLimitCheck", drpcEncoding_File_coordinator_coordinatorproto_protos_coordinator_proto{}, in, out)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+type DRPCCoordinatorServer interface {
+	SpaceSign(context.Context, *SpaceSignRequest) (*SpaceSignResponse, error)
+	FileLimitCheck(context.Context, *FileLimitCheckRequest) (*FileLimitCheckResponse, error)
+}
+
+type DRPCCoordinatorUnimplementedServer struct{}
+
+func (s *DRPCCoordinatorUnimplementedServer) SpaceSign(context.Context, *SpaceSignRequest) (*SpaceSignResponse, error) {
+	return nil, drpcerr.WithCode(errors.New("Unimplemented"), drpcerr.Unimplemented)
+}
+
+func (s *DRPCCoordinatorUnimplementedServer) FileLimitCheck(context.Context, *FileLimitCheckRequest) (*FileLimitCheckResponse, error) {
+	return nil, drpcerr.WithCode(errors.New("Unimplemented"), drpcerr.Unimplemented)
+}
+
+type DRPCCoordinatorDescription struct{}
+
+func (DRPCCoordinatorDescription) NumMethods() int { return 2 }
+
+func (DRPCCoordinatorDescription) Method(n int) (string, drpc.Encoding, drpc.Receiver, interface{}, bool) {
+	switch n {
+	case 0:
+		return "/coordinator.Coordinator/SpaceSign", drpcEncoding_File_coordinator_coordinatorproto_protos_coordinator_proto{},
+			func(srv interface{}, ctx context.Context, in1, in2 interface{}) (drpc.Message, error) {
+				return srv.(DRPCCoordinatorServer).
+					SpaceSign(
+						ctx,
+						in1.(*SpaceSignRequest),
+					)
+			}, DRPCCoordinatorServer.SpaceSign, true
+	case 1:
+		return "/coordinator.Coordinator/FileLimitCheck", drpcEncoding_File_coordinator_coordinatorproto_protos_coordinator_proto{},
+			func(srv interface{}, ctx context.Context, in1, in2 interface{}) (drpc.Message, error) {
+				return srv.(DRPCCoordinatorServer).
+					FileLimitCheck(
+						ctx,
+						in1.(*FileLimitCheckRequest),
+					)
+			}, DRPCCoordinatorServer.FileLimitCheck, true
+	default:
+		return "", nil, nil, nil, false
+	}
+}
+
+func DRPCRegisterCoordinator(mux drpc.Mux, impl DRPCCoordinatorServer) error {
+	return mux.Register(impl, DRPCCoordinatorDescription{})
+}
+
+type DRPCCoordinator_SpaceSignStream interface {
+	drpc.Stream
+	SendAndClose(*SpaceSignResponse) error
+}
+
+type drpcCoordinator_SpaceSignStream struct {
+	drpc.Stream
+}
+
+func (x *drpcCoordinator_SpaceSignStream) SendAndClose(m *SpaceSignResponse) error {
+	if err := x.MsgSend(m, drpcEncoding_File_coordinator_coordinatorproto_protos_coordinator_proto{}); err != nil {
+		return err
+	}
+	return x.CloseSend()
+}
+
+type DRPCCoordinator_FileLimitCheckStream interface {
+	drpc.Stream
+	SendAndClose(*FileLimitCheckResponse) error
+}
+
+type drpcCoordinator_FileLimitCheckStream struct {
+	drpc.Stream
+}
+
+func (x *drpcCoordinator_FileLimitCheckStream) SendAndClose(m *FileLimitCheckResponse) error {
+	if err := x.MsgSend(m, drpcEncoding_File_coordinator_coordinatorproto_protos_coordinator_proto{}); err != nil {
+		return err
+	}
+	return x.CloseSend()
+}

coordinator/coordinatorproto/protos/coordinator.proto

@@ -0,0 +1,55 @@
+syntax = "proto3";
+package coordinator;
+
+option go_package = "coordinator/coordinatorproto";
+
+service Coordinator {
+    // SpaceSign signs a space creation operation
+    rpc SpaceSign(SpaceSignRequest) returns (SpaceSignResponse);
+
+    // FileLimitCheck checks a limit by account and space
+    // can be used only:
+    //  - if a handshake identity matches a given identity
+    //  - if a requester contains in nodeconf list
+    rpc FileLimitCheck(FileLimitCheckRequest) returns (FileLimitCheckResponse);
+}
+
+message SpaceSignRequest {
+    string spaceId = 1;
+}
+
+message SpaceSignResponse {
+    SpaceReceiptWithSignature receipt = 1;
+}
+
+// SpaceReceiptWithSignature contains protobuf encoded receipt and its signature
+message SpaceReceiptWithSignature {
+    bytes spaceReceiptPayload = 1;
+    bytes signature = 2;
+}
+
+// SpaceReceipt contains permission to SpacePush operation
+message SpaceReceipt {
+    // spaceId
+    string spaceId = 1;
+    // peerId of receipt requester
+    string peerId = 2;
+    // identity of space owner
+    bytes accountIdentity = 3;
+    // identity of control node
+    bytes controlNodeIdentity = 4;
+    // unix-timestamp with a deadline time of receipt validity
+    uint64 validUntil = 5;
+}
+
+// FileLimitCheckRequest contains an account identity and spaceId
+// control node checks that identity owns a given space
+message FileLimitCheckRequest {
+    bytes accountIdentity = 1;
+    string spaceId = 2;
+}
+
+// FileLimitCheckResponse returns a current space limit in bytes
+message FileLimitCheckResponse {
+    uint64 limit = 1;
+}

nodeconf/config.go

@@ -6,6 +6,8 @@ const (
 	NodeTypeTree      NodeType = "tree"
 	NodeTypeConsensus NodeType = "consensus"
 	NodeTypeFile      NodeType = "file"
+
+	NodeTypeCoordinator NodeType = "coordinator"
 )
 
 type configGetter interface {
@@ -15,8 +17,7 @@ type configGetter interface {
 type NodeConfig struct {
 	PeerId        string     `yaml:"peerId"`
 	Addresses     []string   `yaml:"address"`
-	SigningKey    string     `yaml:"signingKey,omitempty"`
-	EncryptionKey string     `yaml:"encryptionKey,omitempty"`
+	EncryptionKey string     `yaml:"encryptionPubKey,omitempty"`
 	Types         []NodeType `yaml:"types,omitempty"`
 }
 

nodeconf/configuration.go

@@ -17,6 +17,8 @@ type Configuration interface {
 	FilePeers() []string
 	// ConsensusPeers returns list of consensusnodes
 	ConsensusPeers() []string
+	// CoordinatorPeers returns list of coordinator nodes
+	CoordinatorPeers() []string
 	// Addresses returns map[peerId][]addr with connection addresses for all known nodes
 	Addresses() map[string][]string
 	// CHash returns nodes consistent table
@@ -28,12 +30,13 @@ type Configuration interface {
 }
 
 type configuration struct {
-	id             string
-	accountId      string
-	filePeers      []string
-	consensusPeers []string
-	chash          chash.CHash
-	allMembers     []NodeConfig
+	id               string
+	accountId        string
+	filePeers        []string
+	consensusPeers   []string
+	coordinatorPeers []string
+	chash            chash.CHash
+	allMembers       []NodeConfig
 }
 
 func (c *configuration) Id() string {
@@ -68,6 +71,10 @@ func (c *configuration) ConsensusPeers() []string {
 	return c.consensusPeers
 }
 
+func (c *configuration) CoordinatorPeers() []string {
+	return c.coordinatorPeers
+}
+
 func (c *configuration) Addresses() map[string][]string {
 	res := make(map[string][]string)
 	for _, m := range c.allMembers {

nodeconf/service.go

@@ -8,6 +8,7 @@ import (
 	"github.com/anytypeio/any-sync/util/keys/asymmetric/encryptionkey"
 	"github.com/anytypeio/any-sync/util/keys/asymmetric/signingkey"
 	"github.com/anytypeio/go-chash"
+	"github.com/libp2p/go-libp2p/core/peer"
 )
 
 const CName = "common.nodeconf"
@@ -80,6 +81,9 @@ func (s *service) Init(a *app.App) (err error) {
 		if n.HasType(NodeTypeFile) {
 			fileConfig.filePeers = append(fileConfig.filePeers, n.PeerId)
 		}
+		if n.HasType(NodeTypeCoordinator) {
+			fileConfig.coordinatorPeers = append(fileConfig.coordinatorPeers, n.PeerId)
+		}
 		fileConfig.allMembers = append(fileConfig.allMembers, n)
 	}
 	if err = fileConfig.chash.AddMembers(members...); err != nil {
@@ -103,17 +107,28 @@ func (s *service) GetById(id string) Configuration {
 }
 
 func nodeFromConfigNode(n NodeConfig) (*Node, error) {
-	decodedSigningKey, err := keys.DecodeKeyFromString(
-		n.SigningKey,
-		signingkey.UnmarshalEd25519PrivateKey,
-		nil)
+	p, err := peer.Decode(n.PeerId)
+	if err != nil {
+		return nil, err
+	}
+	ic, err := p.ExtractPublicKey()
 	if err != nil {
 		return nil, err
 	}
 
-	decodedEncryptionKey, err := keys.DecodeKeyFromString(
+	icRaw, err := ic.Raw()
+	if err != nil {
+		return nil, err
+	}
+
+	sigPubKey, err := signingkey.UnmarshalEd25519PublicKey(icRaw)
+	if err != nil {
+		return nil, err
+	}
+
+	encPubKey, err := keys.DecodeKeyFromString(
 		n.EncryptionKey,
-		encryptionkey.NewEncryptionRsaPrivKeyFromBytes,
+		encryptionkey.NewEncryptionRsaPubKeyFromBytes,
 		nil)
 	if err != nil {
 		return nil, err
@@ -122,7 +137,7 @@ func nodeFromConfigNode(n NodeConfig) (*Node, error) {
 	return &Node{
 		Addresses:     n.Addresses,
 		PeerId:        n.PeerId,
-		SigningKey:    decodedSigningKey.GetPublic(),
-		EncryptionKey: decodedEncryptionKey.GetPublic(),
+		SigningKey:    sigPubKey,
+		EncryptionKey: encPubKey,
 	}, nil
 }

testutil/accounttest/accountservice.go

@@ -62,18 +62,13 @@ func (s *AccountTestService) Account() *accountdata.AccountData {
 }
 
 func (s *AccountTestService) NodeConf(addrs []string) nodeconf.NodeConfig {
-	encSk, err := keys.EncodeKeyToString(s.acc.SignKey)
-	if err != nil {
-		panic(err)
-	}
-	encEk, err := keys.EncodeKeyToString(s.acc.EncKey)
+	encEk, err := keys.EncodeKeyToString(s.acc.EncKey.GetPublic())
 	if err != nil {
 		panic(err)
 	}
 	return nodeconf.NodeConfig{
 		PeerId:        s.acc.PeerId,
 		Addresses:     addrs,
-		SigningKey:    encSk,
 		EncryptionKey: encEk,
 		Types:         []nodeconf.NodeType{nodeconf.NodeTypeTree},
 	}