node: expose in-proc RPC (CallRPC), closes #144

cmd/statusd/library.go

@@ -48,6 +48,12 @@ func ResetChainData() *C.char {
 	return makeJSONResponse(err)
 }
 
+//export CallRPC
+func CallRPC(inputJSON *C.char) *C.char {
+	outputJSON := statusAPI.CallRPC(C.GoString(inputJSON))
+	return C.CString(outputJSON)
+}
+
 //export ResumeNode
 func ResumeNode() *C.char {
 	err := fmt.Errorf("%v: %v", common.ErrDeprecatedMethod.Error(), "ResumeNode")

cmd/statusd/utils.go

@@ -55,6 +55,10 @@ func testExportedAPI(t *testing.T, done chan struct{}) {
 			"stop/resume node",
 			testStopResumeNode,
 		},
+		{
+			"call RPC on in-proc handler",
+			testCallRPC,
+		},
 		{
 			"create main and child accounts",
 			testCreateChildAccount,
@@ -371,6 +375,18 @@ func testStopResumeNode(t *testing.T) bool {
 	return true
 }
 
+func testCallRPC(t *testing.T) bool {
+	expected := `{"jsonrpc":"2.0","id":64,"result":"0x47173285a8d7341e5e972fc677286384f802f8ef42a5ec5f03bbfa254cb01fad"}` + "\n"
+	rawResponse := CallRPC(C.CString(`{"jsonrpc":"2.0","method":"web3_sha3","params":["0x68656c6c6f20776f726c64"],"id":64}`))
+	received := C.GoString(rawResponse)
+	if expected != received {
+		t.Errorf("unexpected reponse: expected: %v, got: %v", expected, received)
+		return false
+	}
+
+	return true
+}
+
 func testCreateChildAccount(t *testing.T) bool {
 	// to make sure that we start with empty account (which might get populated during previous tests)
 	if err := statusAPI.Logout(); err != nil {

geth/api/api.go

@@ -97,6 +97,11 @@ func (api *StatusAPI) ResetChainDataAsync() (<-chan struct{}, error) {
 	return api.b.ResetChainData()
 }
 
+// CallRPC executes RPC request on node's in-proc RPC server
+func (api *StatusAPI) CallRPC(inputJSON string) string {
+	return api.b.CallRPC(inputJSON)
+}
+
 // CreateAccount creates an internal geth account
 // BIP44-compatible keys are generated: CKD#1 is stored as account key, CKD#2 stored as sub-account root
 // Public key of CKD#1 is returned, with CKD#2 securely encoded into account key file (to be used for
@@ -156,19 +161,19 @@ func (api *StatusAPI) DiscardTransactions(ids string) map[string]common.RawDisca
 	return api.b.DiscardTransactions(ids)
 }
 
-// Parse creates a new jail cell context, with the given chatID as identifier.
+// JailParse creates a new jail cell context, with the given chatID as identifier.
 // New context executes provided JavaScript code, right after the initialization.
 func (api *StatusAPI) JailParse(chatID string, js string) string {
 	return api.b.jailManager.Parse(chatID, js)
 }
 
-// Call executes given JavaScript function w/i a jail cell context identified by the chatID.
+// JailCall executes given JavaScript function w/i a jail cell context identified by the chatID.
 // Jail cell is clonned before call is executed i.e. all calls execute w/i their own contexts.
 func (api *StatusAPI) JailCall(chatID string, path string, args string) string {
 	return api.b.jailManager.Call(chatID, path, args)
 }
 
-// BaseJS allows to setup initial JavaScript to be loaded on each jail.Parse()
+// JailBaseJS allows to setup initial JavaScript to be loaded on each jail.Parse()
 func (api *StatusAPI) JailBaseJS(js string) {
 	api.b.jailManager.BaseJS(js)
 }

geth/api/backend.go

@@ -20,6 +20,7 @@ type StatusBackend struct {
 	accountManager common.AccountManager
 	txQueueManager common.TxQueueManager
 	jailManager    common.JailManager
+	rpcManager     common.RPCManager
 }
 
 // NewStatusBackend create a new NewStatusBackend instance
@@ -33,6 +34,7 @@ func NewStatusBackend() *StatusBackend {
 		accountManager: accountManager,
 		txQueueManager: node.NewTxQueueManager(nodeManager, accountManager),
 		jailManager:    jail.New(nodeManager),
+		rpcManager:     node.NewRPCManager(nodeManager),
 	}
 }
 
@@ -164,6 +166,11 @@ func (m *StatusBackend) ResetChainData() (<-chan struct{}, error) {
 	return m.nodeReady, err
 }
 
+// CallRPC executes RPC request on node's in-proc RPC server
+func (m *StatusBackend) CallRPC(inputJSON string) string {
+	return m.rpcManager.Call(inputJSON)
+}
+
 // CompleteTransaction instructs backend to complete sending of a given transaction
 func (m *StatusBackend) CompleteTransaction(id, password string) (gethcommon.Hash, error) {
 	return m.txQueueManager.CompleteTransaction(id, password)

geth/api/backend_test.go

@@ -148,6 +148,94 @@ func (s *BackendTestSuite) TestNodeStartStop() {
 	require.True(s.backend.IsNodeRunning())
 }
 
+func (s *BackendTestSuite) TestCallRPC() {
+	require := s.Require()
+	require.NotNil(s.backend)
+
+	nodeConfig, err := MakeTestNodeConfig(params.RinkebyNetworkID)
+	require.NoError(err)
+
+	nodeStarted, err := s.backend.StartNode(nodeConfig)
+	require.NoError(err)
+	require.NotNil(nodeStarted)
+	defer s.backend.StopNode()
+	<-nodeStarted
+
+	progress := make(chan struct{}, 25)
+	type rpcCall struct {
+		inputJSON string
+		validator func(resultJSON string)
+	}
+	var rpcCalls = []rpcCall{
+		{
+			`{"jsonrpc":"2.0","method":"eth_sendTransaction","params":[{
+				"from": "0xb60e8dd61c5d32be8058bb8eb970870f07233155",
+				"to": "0xd46e8dd67c5d32be8058bb8eb970870f07244567",
+				"gas": "0x76c0",
+				"gasPrice": "0x9184e72a000",
+				"value": "0x9184e72a",
+				"data": "0xd46e8dd67c5d32be8d46e8dd67c5d32be8058bb8eb970870f072445675058bb8eb970870f072445675"}],"id":1}`,
+			func(resultJSON string) {
+				log.Info("eth_sendTransaction")
+				s.T().Log("GOT: ", resultJSON)
+				progress <- struct{}{}
+			},
+		},
+		{
+			`{"jsonrpc":"2.0","method":"shh_version","params":[],"id":67}`,
+			func(resultJSON string) {
+				expected := `{"jsonrpc":"2.0","id":67,"result":"0x5"}` + "\n"
+				s.Equal(expected, resultJSON)
+				s.T().Log("shh_version: ", resultJSON)
+				progress <- struct{}{}
+			},
+		},
+		{
+			`{"jsonrpc":"2.0","method":"web3_sha3","params":["0x68656c6c6f20776f726c64"],"id":64}`,
+			func(resultJSON string) {
+				expected := `{"jsonrpc":"2.0","id":64,"result":"0x47173285a8d7341e5e972fc677286384f802f8ef42a5ec5f03bbfa254cb01fad"}` + "\n"
+				s.Equal(expected, resultJSON)
+				s.T().Log("web3_sha3: ", resultJSON)
+				progress <- struct{}{}
+			},
+		},
+		{
+			`{"jsonrpc":"2.0","method":"net_version","params":[]}`, // w/o id
+			func(resultJSON string) {
+				expected := `{"jsonrpc":"2.0","id":1,"result":"4"}` + "\n"
+				s.Equal(expected, resultJSON)
+				s.T().Log("net_version: ", resultJSON)
+				progress <- struct{}{}
+			},
+		},
+		{
+			`{"jsonrpc":"2.0","method":"net_version","params":[],"id":67}`,
+			func(resultJSON string) {
+				expected := `{"jsonrpc":"2.0","id":67,"result":"4"}` + "\n"
+				s.Equal(expected, resultJSON)
+				s.T().Log("net_version: ", resultJSON)
+				progress <- struct{}{}
+			},
+		},
+	}
+
+	cnt := len(rpcCalls) - 1 // send transaction blocks up until complete/discarded/times out
+	for _, r := range rpcCalls {
+		go func(r rpcCall) {
+			s.T().Logf("Run test: %v", r.inputJSON)
+			resultJSON := s.backend.CallRPC(r.inputJSON)
+			r.validator(resultJSON)
+		}(r)
+	}
+
+	for range progress {
+		cnt -= 1
+		if cnt <= 0 {
+			break
+		}
+	}
+}
+
 func (s *BackendTestSuite) TestRaceConditions() {
 	require := s.Require()
 	require.NotNil(s.backend)

geth/common/types.go

@@ -18,6 +18,7 @@ import (
 	"github.com/status-im/status-go/static"
 )
 
+// errors
 var (
 	ErrDeprecatedMethod = errors.New("Method is depricated and will be removed in future release")
 )
@@ -83,6 +84,9 @@ type NodeManager interface {
 
 	// RPCClient exposes reference to RPC client connected to the running node
 	RPCClient() (*rpc.Client, error)
+
+	// RPCServer exposes reference to running node's in-proc RPC server/handler
+	RPCServer() (*rpc.Server, error)
 }
 
 // AccountManager defines expected methods for managing Status accounts
@@ -129,6 +133,12 @@ type AccountManager interface {
 	AddressToDecryptedAccount(address, password string) (accounts.Account, *keystore.Key, error)
 }
 
+// RPCManager defines expected methods for managing RPC client/server
+type RPCManager interface {
+	// Call executes RPC request on node's in-proc RPC server
+	Call(inputJSON string) string
+}
+
 // RawCompleteTransactionResult is a JSON returned from transaction complete function (used internally)
 type RawCompleteTransactionResult struct {
 	Hash  common.Hash

geth/node/manager.go

@@ -27,7 +27,8 @@ var (
 	ErrInvalidLightEthereumService = errors.New("LES service is unavailable")
 	ErrInvalidAccountManager       = errors.New("could not retrieve account manager")
 	ErrAccountKeyStoreMissing      = errors.New("account key store is not set")
-	ErrInvalidRPCClient            = errors.New("RPC service is unavailable")
+	ErrInvalidRPCClient            = errors.New("RPC client is unavailable")
+	ErrInvalidRPCServer            = errors.New("RPC server is unavailable")
 )
 
 // NodeManager manages Status node (which abstracts contained geth node)
@@ -40,6 +41,7 @@ type NodeManager struct {
 	whisperService *whisper.Whisper   // reference to Whisper service
 	lesService     *les.LightEthereum // reference to LES service
 	rpcClient      *rpc.Client        // reference to RPC client
+	rpcServer      *rpc.Server        // reference to RPC server
 }
 
 // NewNodeManager makes new instance of node manager
@@ -157,6 +159,7 @@ func (m *NodeManager) stopNode() (<-chan struct{}, error) {
 		m.lesService = nil
 		m.whisperService = nil
 		m.rpcClient = nil
+		m.rpcServer = nil
 		m.nodeStarted = nil
 		m.node = nil
 		m.Unlock()
@@ -531,3 +534,34 @@ func (m *NodeManager) RPCClient() (*rpc.Client, error) {
 
 	return m.rpcClient, nil
 }
+
+// RPCServer exposes reference to running node's in-proc RPC server/handler
+func (m *NodeManager) RPCServer() (*rpc.Server, error) {
+	if m == nil {
+		return nil, ErrInvalidNodeManager
+	}
+
+	m.RLock()
+	defer m.RUnlock()
+
+	// make sure that node is fully started
+	if m.node == nil || m.nodeStarted == nil {
+		return nil, ErrNoRunningNode
+	}
+	<-m.nodeStarted
+
+	if m.rpcServer == nil {
+		var err error
+		m.rpcServer, err = m.node.InProcRPC()
+		if err != nil {
+			log.Error("Cannot expose on-proc RPC server", "error", err)
+			return nil, ErrInvalidRPCServer
+		}
+	}
+
+	if m.rpcServer == nil {
+		return nil, ErrInvalidRPCServer
+	}
+
+	return m.rpcServer, nil
+}

geth/node/manager_test.go

@@ -148,6 +148,13 @@ func (s *ManagerTestSuite) TestReferences() {
 			},
 			node.ErrInvalidNodeManager,
 		},
+		{
+			"null manager, get RPC Server",
+			func() (interface{}, error) {
+				return nilNodeManager.RPCServer()
+			},
+			node.ErrInvalidNodeManager,
+		},
 		{
 			"non-null manager, no running node, RestartNode()",
 			func() (interface{}, error) {
@@ -225,6 +232,13 @@ func (s *ManagerTestSuite) TestReferences() {
 			},
 			node.ErrNoRunningNode,
 		},
+		{
+			"non-null manager, no running node, get RPC Server",
+			func() (interface{}, error) {
+				return s.NodeManager.RPCServer()
+			},
+			node.ErrNoRunningNode,
+		},
 	}
 	for _, testCase := range noNodeTests {
 		s.T().Log(testCase.name)
@@ -290,6 +304,13 @@ func (s *ManagerTestSuite) TestReferences() {
 			},
 			&rpc.Client{},
 		},
+		{
+			"node is running, get RPC Server",
+			func() (interface{}, error) {
+				return s.NodeManager.RPCServer()
+			},
+			&rpc.Server{},
+		},
 	}
 	for _, testCase := range nodeReadyTestCases {
 		obj, err := testCase.initFn()
@@ -507,6 +528,12 @@ func (s *ManagerTestSuite) TestRaceConditions() {
 			s.T().Logf("RPCClient(), error: %v", err)
 			progress <- struct{}{}
 		},
+		func(config *params.NodeConfig) {
+			log.Info("RPCServer()")
+			_, err := s.NodeManager.RPCServer()
+			s.T().Logf("RPCServer(), error: %v", err)
+			progress <- struct{}{}
+		},
 	}
 
 	// increase StartNode()/StopNode() population

geth/node/rpc.go

@@ -0,0 +1,143 @@
+package node
+
+import (
+	"encoding/json"
+	"errors"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/ethereum/go-ethereum/les/status"
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/status-im/status-go/geth/common"
+)
+
+const (
+	jsonrpcVersion         = "2.0"
+	serviceMethodSeparator = "_"
+)
+
+type jsonRequest struct {
+	Method  string          `json:"method"`
+	Version string          `json:"jsonrpc"`
+	ID      int             `json:"id,omitempty"`
+	Payload json.RawMessage `json:"params,omitempty"`
+}
+
+type jsonError struct {
+	Code    int         `json:"code"`
+	Message string      `json:"message"`
+	Data    interface{} `json:"data,omitempty"`
+}
+
+type jsonErrResponse struct {
+	Version string      `json:"jsonrpc"`
+	ID      interface{} `json:"id,omitempty"`
+	Error   jsonError   `json:"error"`
+}
+
+// RPCManager abstract RPC management API (for both client and server)
+type RPCManager struct {
+	sync.Mutex
+	requestID   int
+	nodeManager common.NodeManager
+}
+
+// errors
+var (
+	ErrInvalidMethod       = errors.New("method does not exist")
+	ErrRPCServerTimeout    = errors.New("RPC server cancelled call due to timeout")
+	ErrRPCServerCallFailed = errors.New("RPC server cannot complete request")
+)
+
+// NewRPCManager returns new instance of RPC client
+func NewRPCManager(nodeManager common.NodeManager) *RPCManager {
+	return &RPCManager{
+		nodeManager: nodeManager,
+	}
+}
+
+// Call executes RPC request on node's in-proc RPC server
+func (c *RPCManager) Call(inputJSON string) string {
+	server, err := c.nodeManager.RPCServer()
+	if err != nil {
+		return c.makeJSONErrorResponse(err)
+	}
+
+	// allow HTTP requests to block w/o
+	outputJSON := make(chan string, 1)
+	go func() {
+		inputJSON, err = c.prepare(inputJSON)
+		if err != nil {
+			outputJSON <- c.makeJSONErrorResponse(err)
+			return
+		}
+
+		httpReq := httptest.NewRequest("POST", "/", strings.NewReader(inputJSON))
+		rr := httptest.NewRecorder()
+		server.ServeHTTP(rr, httpReq)
+
+		// Check the status code is what we expect.
+		if respStatus := rr.Code; respStatus != http.StatusOK {
+			log.Error("handler returned wrong status code", "got", respStatus, "want", http.StatusOK)
+			outputJSON <- c.makeJSONErrorResponse(ErrRPCServerCallFailed)
+			return
+		}
+
+		// everything is ok, return
+		outputJSON <- rr.Body.String()
+	}()
+
+	// wait till call is complete
+	select {
+	case out := <-outputJSON:
+		return out
+	case <-time.After((status.DefaultTxSendCompletionTimeout + 10) * time.Minute): // give up eventually
+		// pass
+	}
+
+	return c.makeJSONErrorResponse(ErrRPCServerTimeout)
+}
+
+// prepare applies necessary transformations to incoming JSON
+func (c *RPCManager) prepare(inputJSON string) (string, error) {
+	var in jsonRequest
+	if err := json.Unmarshal(json.RawMessage(inputJSON), &in); err != nil {
+		return inputJSON, err
+	}
+
+	elems := strings.Split(in.Method, serviceMethodSeparator)
+	if len(elems) != 2 {
+		return inputJSON, ErrInvalidMethod
+	}
+
+	// inject next ID
+	if in.ID == 0 {
+		c.Lock()
+		c.requestID++
+		c.Unlock()
+		in.ID = c.requestID
+	}
+
+	outputJSON, err := json.Marshal(&in)
+	if err != nil {
+		return inputJSON, err
+	}
+
+	return string(outputJSON), nil
+}
+
+// makeJSONErrorResponse returns error as JSON response
+func (c *RPCManager) makeJSONErrorResponse(err error) string {
+	response := jsonErrResponse{
+		Version: jsonrpcVersion,
+		Error: jsonError{
+			Message: err.Error(),
+		},
+	}
+
+	outBytes, _ := json.Marshal(&response)
+	return string(outBytes)
+}

geth/node/rpc_test.go

@@ -0,0 +1,120 @@
+package node_test
+
+import (
+	"testing"
+
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/status-im/status-go/geth/node"
+	"github.com/status-im/status-go/geth/params"
+	. "github.com/status-im/status-go/geth/testing"
+	"github.com/stretchr/testify/suite"
+)
+
+func TestRPCTestSuite(t *testing.T) {
+	suite.Run(t, new(RPCTestSuite))
+}
+
+type RPCTestSuite struct {
+	BaseTestSuite
+}
+
+func (s *RPCTestSuite) SetupTest() {
+	require := s.Require()
+	s.NodeManager = node.NewNodeManager()
+	require.NotNil(s.NodeManager)
+	require.IsType(&node.NodeManager{}, s.NodeManager)
+}
+
+func (s *RPCTestSuite) TestCallRPC() {
+	require := s.Require()
+	require.NotNil(s.NodeManager)
+
+	rpcClient := node.NewRPCManager(s.NodeManager)
+	require.NotNil(rpcClient)
+
+	nodeConfig, err := MakeTestNodeConfig(params.RinkebyNetworkID)
+	require.NoError(err)
+
+	nodeConfig.IPCEnabled = false
+	nodeConfig.WSEnabled = false
+	nodeConfig.HTTPHost = "" // to make sure that no HTTP interface is started
+	nodeStarted, err := s.NodeManager.StartNode(nodeConfig)
+	require.NoError(err)
+	require.NotNil(nodeConfig)
+	defer s.NodeManager.StopNode()
+	<-nodeStarted
+
+	progress := make(chan struct{}, 25)
+	type rpcCall struct {
+		inputJSON string
+		validator func(resultJSON string)
+	}
+	var rpcCalls = []rpcCall{
+		{
+			`{"jsonrpc":"2.0","method":"eth_sendTransaction","params":[{
+				"from": "0xb60e8dd61c5d32be8058bb8eb970870f07233155",
+				"to": "0xd46e8dd67c5d32be8058bb8eb970870f07244567",
+				"gas": "0x76c0",
+				"gasPrice": "0x9184e72a000",
+				"value": "0x9184e72a",
+				"data": "0xd46e8dd67c5d32be8d46e8dd67c5d32be8058bb8eb970870f072445675058bb8eb970870f072445675"}],"id":1}`,
+			func(resultJSON string) {
+				log.Info("eth_sendTransaction")
+				s.T().Log("GOT: ", resultJSON)
+				progress <- struct{}{}
+			},
+		},
+		{
+			`{"jsonrpc":"2.0","method":"shh_version","params":[],"id":67}`,
+			func(resultJSON string) {
+				expected := `{"jsonrpc":"2.0","id":67,"result":"0x5"}` + "\n"
+				s.Equal(expected, resultJSON)
+				s.T().Log("shh_version: ", resultJSON)
+				progress <- struct{}{}
+			},
+		},
+		{
+			`{"jsonrpc":"2.0","method":"web3_sha3","params":["0x68656c6c6f20776f726c64"],"id":64}`,
+			func(resultJSON string) {
+				expected := `{"jsonrpc":"2.0","id":64,"result":"0x47173285a8d7341e5e972fc677286384f802f8ef42a5ec5f03bbfa254cb01fad"}` + "\n"
+				s.Equal(expected, resultJSON)
+				s.T().Log("web3_sha3: ", resultJSON)
+				progress <- struct{}{}
+			},
+		},
+		{
+			`{"jsonrpc":"2.0","method":"net_version","params":[]}`, // w/o id
+			func(resultJSON string) {
+				expected := `{"jsonrpc":"2.0","id":1,"result":"4"}` + "\n"
+				s.Equal(expected, resultJSON)
+				s.T().Log("net_version: ", resultJSON)
+				progress <- struct{}{}
+			},
+		},
+		{
+			`{"jsonrpc":"2.0","method":"net_version","params":[],"id":67}`,
+			func(resultJSON string) {
+				expected := `{"jsonrpc":"2.0","id":67,"result":"4"}` + "\n"
+				s.Equal(expected, resultJSON)
+				s.T().Log("net_version: ", resultJSON)
+				progress <- struct{}{}
+			},
+		},
+	}
+
+	cnt := len(rpcCalls) - 1 // send transaction blocks up until complete/discarded/times out
+	for _, r := range rpcCalls {
+		go func(r rpcCall) {
+			s.T().Logf("Run test: %v", r.inputJSON)
+			resultJSON := rpcClient.Call(r.inputJSON)
+			r.validator(resultJSON)
+		}(r)
+	}
+
+	for range progress {
+		cnt -= 1
+		if cnt <= 0 {
+			break
+		}
+	}
+}

vendor/github.com/ethereum/go-ethereum/node/node.go

@@ -568,6 +568,18 @@ func (n *Node) Attach() (*rpc.Client, error) {
 	return rpc.DialInProc(n.inprocHandler), nil
 }
 
+// InProcRPC exposes in-proc RPC server
+func (n *Node) InProcRPC() (*rpc.Server, error) {
+	n.lock.RLock()
+	defer n.lock.RUnlock()
+
+	if n.server == nil || n.inprocHandler == nil {
+		return nil, ErrNodeStopped
+	}
+
+	return n.inprocHandler, nil
+}
+
 // Server retrieves the currently running P2P network layer. This method is meant
 // only to inspect fields of the currently running server, life cycle management
 // should be left to this Node entity.