From 69c2e537c677c9c1314df74aa035f0b9d178edb7 Mon Sep 17 00:00:00 2001 From: Lewis Marshall Date: Sun, 28 May 2017 19:39:06 +0100 Subject: [PATCH 1/5] p2p/simulations: Use WebSocket RPC Signed-off-by: Lewis Marshall --- node/api.go | 2 +- node/config.go | 7 + node/node.go | 8 +- p2p/simulations/adapters/exec.go | 112 +++++++++----- p2p/simulations/adapters/inproc.go | 37 +++-- p2p/simulations/adapters/rpc_mux.go | 218 ---------------------------- 6 files changed, 106 insertions(+), 278 deletions(-) delete mode 100644 p2p/simulations/adapters/rpc_mux.go diff --git a/node/api.go b/node/api.go index 1b2ef2ed90..1b04b70938 100644 --- a/node/api.go +++ b/node/api.go @@ -203,7 +203,7 @@ func (api *PrivateAdminAPI) StartWS(host *string, port *int, allowedOrigins *str } } - if err := api.node.startWS(fmt.Sprintf("%s:%d", *host, *port), api.node.rpcAPIs, modules, origins); err != nil { + if err := api.node.startWS(fmt.Sprintf("%s:%d", *host, *port), api.node.rpcAPIs, modules, origins, api.node.config.WSExposeAll); err != nil { return false, err } return true, nil diff --git a/node/config.go b/node/config.go index 61e0008ef9..7021eaadf0 100644 --- a/node/config.go +++ b/node/config.go @@ -128,6 +128,13 @@ type Config struct { // If the module list is empty, all RPC API endpoints designated public will be // exposed. WSModules []string `toml:",omitempty"` + + // WSExposeAll exposes all API modules via the WebSocket RPC interface rather + // than just the public ones. + // + // *WARNING* Only set this if the node is running in a trusted network, exposing + // private APIs to untrusted users is a major security risk. + WSExposeAll bool `toml:",omitempty"` } // IPCEndpoint resolves an IPC endpoint based on a configured value, taking into diff --git a/node/node.go b/node/node.go index a89110599f..854efd28cb 100644 --- a/node/node.go +++ b/node/node.go @@ -275,7 +275,7 @@ func (n *Node) startRPC(services map[reflect.Type]Service) error { n.stopInProc() return err } - if err := n.startWS(n.wsEndpoint, apis, n.config.WSModules, n.config.WSOrigins); err != nil { + if err := n.startWS(n.wsEndpoint, apis, n.config.WSModules, n.config.WSOrigins, n.config.WSExposeAll); err != nil { n.stopHTTP() n.stopIPC() n.stopInProc() @@ -426,7 +426,7 @@ func (n *Node) stopHTTP() { } // startWS initializes and starts the websocket RPC endpoint. -func (n *Node) startWS(endpoint string, apis []rpc.API, modules []string, wsOrigins []string) error { +func (n *Node) startWS(endpoint string, apis []rpc.API, modules []string, wsOrigins []string, exposeAll bool) error { // Short circuit if the WS endpoint isn't being exposed if endpoint == "" { return nil @@ -439,7 +439,7 @@ func (n *Node) startWS(endpoint string, apis []rpc.API, modules []string, wsOrig // Register all the APIs exposed by the services handler := rpc.NewServer() for _, api := range apis { - if whitelist[api.Namespace] || (len(whitelist) == 0 && api.Public) { + if exposeAll || whitelist[api.Namespace] || (len(whitelist) == 0 && api.Public) { if err := handler.RegisterName(api.Namespace, api.Service); err != nil { return err } @@ -455,7 +455,7 @@ func (n *Node) startWS(endpoint string, apis []rpc.API, modules []string, wsOrig return err } go rpc.NewWSServer(wsOrigins, handler).Serve(listener) - log.Info(fmt.Sprintf("WebSocket endpoint opened: ws://%s", endpoint)) + log.Info(fmt.Sprintf("WebSocket endpoint opened: ws://%s", listener.Addr())) // All listeners booted successfully n.wsEndpoint = endpoint diff --git a/p2p/simulations/adapters/exec.go b/p2p/simulations/adapters/exec.go index f18f4c96f5..61cd5c5c01 100644 --- a/p2p/simulations/adapters/exec.go +++ b/p2p/simulations/adapters/exec.go @@ -1,6 +1,7 @@ package adapters import ( + "bufio" "context" "crypto/ecdsa" "encoding/json" @@ -12,7 +13,9 @@ import ( "os/exec" "os/signal" "path/filepath" + "regexp" "strings" + "sync" "syscall" "time" @@ -22,6 +25,7 @@ import ( "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p/discover" "github.com/ethereum/go-ethereum/rpc" + "golang.org/x/net/websocket" ) // ExecAdapter is a NodeAdapter which runs nodes by executing the current @@ -69,6 +73,10 @@ func (e *ExecAdapter) NewNode(config *NodeConfig) (Node, error) { Node: config, } conf.Stack.DataDir = filepath.Join(dir, "data") + conf.Stack.WSHost = "127.0.0.1" + conf.Stack.WSPort = 0 + conf.Stack.WSOrigins = []string{"*"} + conf.Stack.WSExposeAll = true conf.Stack.P2P.EnableMsgEvents = false conf.Stack.P2P.NoDiscovery = true conf.Stack.P2P.NAT = nil @@ -101,7 +109,7 @@ type ExecNode struct { Info *p2p.NodeInfo client *rpc.Client - rpcMux *rpcMux + wsAddr string newCmd func() *exec.Cmd key *ecdsa.PrivateKey } @@ -120,6 +128,10 @@ func (n *ExecNode) Client() (*rpc.Client, error) { return n.client, nil } +// wsAddrPattern is a regex used to read the WebSocket address from the node's +// log +var wsAddrPattern = regexp.MustCompile(`ws://[\d.:]+`) + // Start exec's the node passing the ID and service as command line arguments // and the node config encoded as JSON in the _P2P_NODE_CONFIG environment // variable @@ -142,29 +154,61 @@ func (n *ExecNode) Start(snapshots map[string][]byte) (err error) { return fmt.Errorf("error generating node config: %s", err) } - // create a net.Pipe for RPC communication over stdin / stdout - pipe1, pipe2 := net.Pipe() + // use a pipe for stderr so we can both copy the node's stderr to + // os.Stderr and read the WebSocket address from the logs + stderrR, stderrW := io.Pipe() + stderr := io.MultiWriter(os.Stderr, stderrW) // start the node cmd := n.newCmd() - cmd.Stdin = pipe1 - cmd.Stdout = pipe1 - cmd.Stderr = os.Stderr + cmd.Stdout = os.Stdout + cmd.Stderr = stderr cmd.Env = append(os.Environ(), fmt.Sprintf("_P2P_NODE_CONFIG=%s", confData)) if err := cmd.Start(); err != nil { return fmt.Errorf("error starting node: %s", err) } n.Cmd = cmd + // read the WebSocket address from the stderr logs + var wsAddr string + errC := make(chan error) + go func() { + s := bufio.NewScanner(stderrR) + for s.Scan() { + if strings.Contains(s.Text(), "WebSocket endpoint opened:") { + wsAddr = wsAddrPattern.FindString(s.Text()) + break + } + } + select { + case errC <- s.Err(): + default: + } + }() + select { + case err := <-errC: + if err != nil { + return fmt.Errorf("error reading WebSocket address from stderr: %s", err) + } else if wsAddr == "" { + return errors.New("failed to read WebSocket address from stderr") + } + case <-time.After(10 * time.Second): + return errors.New("timed out waiting for WebSocket address on stderr") + } + // create the RPC client and load the node info - n.rpcMux = newRPCMux(pipe2) - n.client = n.rpcMux.Client() ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() + client, err := rpc.DialWebsocket(ctx, wsAddr, "") + if err != nil { + return fmt.Errorf("error dialing rpc websocket: %s", err) + } var info p2p.NodeInfo - if err := n.client.CallContext(ctx, &info, "admin_nodeInfo"); err != nil { + if err := client.CallContext(ctx, &info, "admin_nodeInfo"); err != nil { return fmt.Errorf("error getting node info: %s", err) } + n.client = client + n.wsAddr = wsAddr n.Info = &info return nil @@ -193,6 +237,7 @@ func (n *ExecNode) Stop() error { if n.client != nil { n.client.Close() n.client = nil + n.wsAddr = "" n.Info = nil } @@ -222,13 +267,30 @@ func (n *ExecNode) NodeInfo() *p2p.NodeInfo { return info } -// ServeRPC serves RPC requests over the given connection using the node's -// RPC multiplexer -func (n *ExecNode) ServeRPC(conn net.Conn) error { - if n.rpcMux == nil { - return errors.New("RPC not started") +// ServeRPC serves RPC requests over the given connection by dialling the +// node's WebSocket address and joining the two connections +func (n *ExecNode) ServeRPC(clientConn net.Conn) error { + conn, err := websocket.Dial(n.wsAddr, "", "http://localhost") + if err != nil { + return err } - n.rpcMux.Serve(conn) + var wg sync.WaitGroup + wg.Add(2) + join := func(src, dst net.Conn) { + defer wg.Done() + io.Copy(dst, src) + // close the write end of the destination connection + if cw, ok := dst.(interface { + CloseWrite() error + }); ok { + cw.CloseWrite() + } else { + dst.Close() + } + } + go join(conn, clientConn) + go join(clientConn, conn) + wg.Wait() return nil } @@ -313,15 +375,6 @@ func execP2PNode() { log.Crit("error starting p2p node", "err", err) } - // use stdin / stdout for RPC to avoid the parent needing to access - // either the local filesystem or TCP stack (useful when running in - // Docker) - handler, err := stack.RPCHandler() - if err != nil { - log.Crit("error getting RPC server", "err", err) - } - go handler.ServeCodec(rpc.NewJSONCodec(&stdioConn{os.Stdin, os.Stdout}), rpc.OptionMethodInvocation|rpc.OptionSubscriptions) - go func() { sigc := make(chan os.Signal, 1) signal.Notify(sigc, syscall.SIGTERM) @@ -404,14 +457,3 @@ func (api SnapshotAPI) Snapshot() (map[string][]byte, error) { } return snapshots, nil } - -// stdioConn wraps os.Stdin / os.Stdout with a no-op Close method so we can -// use stdio for RPC messages -type stdioConn struct { - io.Reader - io.Writer -} - -func (r *stdioConn) Close() error { - return nil -} diff --git a/p2p/simulations/adapters/inproc.go b/p2p/simulations/adapters/inproc.go index 66b209dddb..f0a989d829 100644 --- a/p2p/simulations/adapters/inproc.go +++ b/p2p/simulations/adapters/inproc.go @@ -119,7 +119,6 @@ type SimNode struct { node *node.Node running []node.Service client *rpc.Client - rpcMux *rpcMux } // Addr returns the node's discovery address @@ -135,24 +134,28 @@ func (self *SimNode) Node() *discover.Node { // Client returns an rpc.Client which can be used to communicate with the // underlying service (it is set once the node has started) func (self *SimNode) Client() (*rpc.Client, error) { - self.lock.Lock() - defer self.lock.Unlock() + self.lock.RLock() + defer self.lock.RUnlock() if self.client == nil { - return nil, errors.New("RPC not started") + return nil, errors.New("node not started") } return self.client, nil } -// ServeRPC serves RPC requests over the given connection using the node's -// RPC multiplexer +// ServeRPC serves RPC requests over the given connection by creating an +// in-process client to the node's RPC server func (self *SimNode) ServeRPC(conn net.Conn) error { - self.lock.Lock() - mux := self.rpcMux - self.lock.Unlock() - if mux == nil { - return errors.New("RPC not started") + self.lock.RLock() + node := self.node + self.lock.RUnlock() + if node == nil { + return errors.New("node not started") } - mux.Serve(conn) + handler, err := node.RPCHandler() + if err != nil { + return err + } + handler.ServeCodec(rpc.NewJSONCodec(conn), rpc.OptionMethodInvocation|rpc.OptionSubscriptions) return nil } @@ -213,18 +216,12 @@ func (self *SimNode) Start(snapshots map[string][]byte) error { return err } + // create an in-process RPC client handler, err := node.RPCHandler() if err != nil { return err } - - // create an in-process RPC multiplexer - pipe1, pipe2 := net.Pipe() - go handler.ServeCodec(rpc.NewJSONCodec(pipe1), rpc.OptionMethodInvocation|rpc.OptionSubscriptions) - self.rpcMux = newRPCMux(pipe2) - - // create an in-process RPC client - self.client = self.rpcMux.Client() + self.client = rpc.DialInProc(handler) self.node = node diff --git a/p2p/simulations/adapters/rpc_mux.go b/p2p/simulations/adapters/rpc_mux.go deleted file mode 100644 index 6d2c5b58ae..0000000000 --- a/p2p/simulations/adapters/rpc_mux.go +++ /dev/null @@ -1,218 +0,0 @@ -// Copyright 2016 The go-ethereum Authors -// This file is part of the go-ethereum library. -// -// The go-ethereum library is free software: you can redistribute it and/or modify -// it under the terms of the GNU Lesser General Public License as published by -// the Free Software Foundation, either version 3 of the License, or -// (at your option) any later version. -// -// The go-ethereum library is distributed in the hope that it will be useful, -// but WITHOUT ANY WARRANTY; without even the implied warranty of -// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -// GNU Lesser General Public License for more details. -// -// You should have received a copy of the GNU Lesser General Public License -// along with the go-ethereum library. If not, see . - -package adapters - -import ( - "encoding/json" - "net" - "strconv" - "strings" - "sync" - - "github.com/ethereum/go-ethereum/rpc" -) - -// rpcMux is an RPC multiplexer which allows many clients to make RPC requests -// over a single connection by changing each request's ID to a unique value. -// -// This is used by node adapters so that simulations can create many RPC -// clients all sending requests over the underlying node's stdin / stdout. -type rpcMux struct { - conn net.Conn - - mtx sync.Mutex - idCounter uint64 - msgMap map[uint64]*rpcMsg - subMap map[string]*rpcReply - send chan *rpcMsg -} - -type rpcMsg struct { - Method string `json:"method,omitempty"` - Version string `json:"jsonrpc,omitempty"` - ID json.RawMessage `json:"id,omitempty"` - Payload json.RawMessage `json:"params,omitempty"` - Result json.RawMessage `json:"result,omitempty"` - Error json.RawMessage `json:"error,omitempty"` - - id uint64 - reply *rpcReply -} - -// rpcSub is the payload or result of a subscription RPC message -type rpcSub struct { - Subscription string `json:"subscription"` - Result json.RawMessage `json:"result,omitempty"` -} - -// rpcReply receives replies to RPC messages for a particular client -type rpcReply struct { - ch chan *rpcMsg - closeOnce sync.Once -} - -func (r *rpcReply) close() { - r.closeOnce.Do(func() { close(r.ch) }) -} - -func newRPCMux(conn net.Conn) *rpcMux { - mux := &rpcMux{ - msgMap: make(map[uint64]*rpcMsg), - subMap: make(map[string]*rpcReply), - send: make(chan *rpcMsg), - } - go mux.sendLoop(conn) - go mux.recvLoop(conn) - return mux -} - -// Client creates a new RPC client which sends messages through the multiplexer -func (mux *rpcMux) Client() *rpc.Client { - pipe1, pipe2 := net.Pipe() - go mux.Serve(pipe1) - return rpc.NewClientWithConn(pipe2) -} - -// Serve reads RPC messages from the given connection, forwards them to the -// multiplexed connnection and writes replies back to the given connection -func (mux *rpcMux) Serve(conn net.Conn) { - // reply will receive replies to any messages we send - reply := &rpcReply{ch: make(chan *rpcMsg)} - defer func() { - // drain the channel to prevent blocking the recvLoop - for range reply.ch { - } - }() - - // start a goroutine to read RPC messages from the connection and - // forward them to the sendLoop - done := make(chan struct{}) - go func() { - defer close(done) - dec := json.NewDecoder(conn) - for { - msg := &rpcMsg{} - if err := dec.Decode(msg); err != nil { - return - } - msg.reply = reply - mux.send <- msg - } - }() - - // write message replies to the connection - enc := json.NewEncoder(conn) - for { - select { - case msg, ok := <-reply.ch: - if !ok { - return - } - if err := enc.Encode(msg); err != nil { - return - } - case <-done: - return - } - } -} - -// sendLoop receives messages from the send channel, changes their ID and -// writes them to the given connection -func (mux *rpcMux) sendLoop(conn net.Conn) { - enc := json.NewEncoder(conn) - for msg := range mux.send { - if err := enc.Encode(mux.newMsg(msg)); err != nil { - return - } - } -} - -// recvLoop reads messages from the given connection, changes their ID back -// to the oringal value and sends them to the message's reply channel -func (mux *rpcMux) recvLoop(conn net.Conn) { - // close all reply channels if we get an error - defer func() { - mux.mtx.Lock() - defer mux.mtx.Unlock() - for _, msg := range mux.msgMap { - msg.reply.close() - } - }() - - dec := json.NewDecoder(conn) - for { - msg := &rpcMsg{} - if err := dec.Decode(msg); err != nil { - return - } - if reply := mux.lookup(msg); reply != nil { - reply.ch <- msg - } - } -} - -// newMsg copies the given message and changes it's ID to a unique value -func (mux *rpcMux) newMsg(msg *rpcMsg) *rpcMsg { - mux.mtx.Lock() - defer mux.mtx.Unlock() - id := mux.idCounter - mux.idCounter++ - mux.msgMap[id] = msg - newMsg := *msg - newMsg.ID = json.RawMessage(strconv.FormatUint(id, 10)) - return &newMsg -} - -// lookup looks up the original message for which the given message is a reply -func (mux *rpcMux) lookup(msg *rpcMsg) *rpcReply { - mux.mtx.Lock() - defer mux.mtx.Unlock() - - // if the message has no ID, it is a subscription notification so - // lookup the original subscribe message - if msg.ID == nil { - sub := &rpcSub{} - if err := json.Unmarshal(msg.Payload, sub); err != nil { - return nil - } - return mux.subMap[sub.Subscription] - } - - // lookup the original message and restore the ID - id, err := strconv.ParseUint(string(msg.ID), 10, 64) - if err != nil { - return nil - } - origMsg, ok := mux.msgMap[id] - if !ok { - return nil - } - delete(mux.msgMap, id) - msg.ID = origMsg.ID - - // if the original message was a subscription, store the subscription - // ID so we can detect notifications - if strings.HasSuffix(string(origMsg.Method), "_subscribe") { - var result string - if err := json.Unmarshal(msg.Result, &result); err == nil { - mux.subMap[result] = origMsg.reply - } - } - - return origMsg.reply -} From 1341843f7eebf93ed5deeb0f0b81a185cd2239bc Mon Sep 17 00:00:00 2001 From: Lewis Marshall Date: Sun, 28 May 2017 19:39:33 +0100 Subject: [PATCH 2/5] p2p/simulations: Fix HTTP tests Signed-off-by: Lewis Marshall --- p2p/simulations/http_test.go | 30 ++++++++++++------------------ 1 file changed, 12 insertions(+), 18 deletions(-) diff --git a/p2p/simulations/http_test.go b/p2p/simulations/http_test.go index fa55bd29f3..d91127450e 100644 --- a/p2p/simulations/http_test.go +++ b/p2p/simulations/http_test.go @@ -185,8 +185,7 @@ func TestHTTPNetwork(t *testing.T) { // create 2 nodes nodeIDs := make([]string, 2) for i := 0; i < 2; i++ { - config := &adapters.NodeConfig{} - node, err := client.CreateNode(network.ID, config) + node, err := client.CreateNode(network.ID, nil) if err != nil { t.Fatalf("error creating node: %s", err) } @@ -286,8 +285,8 @@ func (t *expectEvents) expect(events ...*Event) { if event.Node == nil { t.Fatal("expected event.Node to be set") } - if event.Node.ID().NodeID != expected.Node.ID().NodeID { - t.Fatalf("expected node event %d to have id %q, got %q", i, expected.Node.ID().Label(), event.Node.ID().Label()) + if event.Node.ID() != expected.Node.ID() { + t.Fatalf("expected node event %d to have id %q, got %q", i, expected.Node.ID().TerminalString(), event.Node.ID().TerminalString()) } if event.Node.Up != expected.Node.Up { t.Fatalf("expected node event %d to have up=%t, got up=%t", i, expected.Node.Up, event.Node.Up) @@ -297,17 +296,11 @@ func (t *expectEvents) expect(events ...*Event) { if event.Conn == nil { t.Fatal("expected event.Conn to be set") } - if event.Conn.One == nil { - t.Fatal("expected event.Conn.One to be set") + if event.Conn.One != expected.Conn.One { + t.Fatalf("expected conn event %d to have one=%q, got one=%q", i, expected.Conn.One.TerminalString(), event.Conn.One.TerminalString()) } - if event.Conn.Other == nil { - t.Fatal("expected event.Conn.Other to be set") - } - if event.Conn.One.NodeID != expected.Conn.One.NodeID { - t.Fatalf("expected conn event %d to have one=%q, got one=%q", i, expected.Conn.One.Label(), event.Conn.One.Label()) - } - if event.Conn.Other.NodeID != expected.Conn.Other.NodeID { - t.Fatalf("expected conn event %d to have other=%q, got other=%q", i, expected.Conn.Other.Label(), event.Conn.Other.Label()) + if event.Conn.Other != expected.Conn.Other { + t.Fatalf("expected conn event %d to have other=%q, got other=%q", i, expected.Conn.Other.TerminalString(), event.Conn.Other.TerminalString()) } if event.Conn.Up != expected.Conn.Up { t.Fatalf("expected conn event %d to have up=%t, got up=%t", i, expected.Conn.Up, event.Conn.Up) @@ -335,7 +328,7 @@ func TestHTTPNodeRPC(t *testing.T) { if err != nil { t.Fatalf("error creating network: %s", err) } - node, err := client.CreateNode(network.ID, &adapters.NodeConfig{}) + node, err := client.CreateNode(network.ID, nil) if err != nil { t.Fatalf("error creating node: %s", err) } @@ -401,7 +394,7 @@ func TestHTTPSnapshot(t *testing.T) { nodeCount := 2 nodes := make([]*p2p.NodeInfo, nodeCount) for i := 0; i < nodeCount; i++ { - node, err := client.CreateNode(network.ID, &adapters.NodeConfig{}) + node, err := client.CreateNode(network.ID, nil) if err != nil { t.Fatalf("error creating node: %s", err) } @@ -435,8 +428,9 @@ func TestHTTPSnapshot(t *testing.T) { t.Fatalf("error creating snapshot: %s", err) } for i, state := range states { - if string(snap.Nodes[i].Snapshot) != state { - t.Fatalf("expected snapshot state %q, got %q", state, snap.Nodes[i].Snapshot) + gotState := snap.Nodes[i].Snapshots["test"] + if string(gotState) != state { + t.Fatalf("expected snapshot state %q, got %q", state, gotState) } } From 177e720b087547440c083eb7cc659936cacdd6f1 Mon Sep 17 00:00:00 2001 From: Lewis Marshall Date: Sun, 28 May 2017 23:16:44 +0100 Subject: [PATCH 3/5] p2p/simulations: Support inter-node RPC connections Signed-off-by: Lewis Marshall --- p2p/simulations/adapters/exec.go | 133 +++++++++++------- p2p/simulations/adapters/inproc.go | 33 ++++- p2p/simulations/adapters/types.go | 14 +- p2p/simulations/examples/connectivity.go | 4 +- p2p/simulations/http_test.go | 8 +- p2p/testing/protocoltester.go | 8 +- .../simulations/discovery/discovery_test.go | 6 +- swarm/network/simulations/overlay.go | 5 +- swarm/pss/pss_test.go | 19 ++- 9 files changed, 148 insertions(+), 82 deletions(-) diff --git a/p2p/simulations/adapters/exec.go b/p2p/simulations/adapters/exec.go index 61cd5c5c01..319e4353a2 100644 --- a/p2p/simulations/adapters/exec.go +++ b/p2p/simulations/adapters/exec.go @@ -36,12 +36,17 @@ import ( // execP2PNode function for more information. type ExecAdapter struct { BaseDir string + + nodes map[discover.NodeID]*ExecNode } // NewExecAdapter returns an ExecAdapter which stores node data in // subdirectories of the given base directory func NewExecAdapter(baseDir string) *ExecAdapter { - return &ExecAdapter{BaseDir: baseDir} + return &ExecAdapter{ + BaseDir: baseDir, + nodes: make(map[discover.NodeID]*ExecNode), + } } // Name returns the name of the adapter for logging purpoeses @@ -86,11 +91,13 @@ func (e *ExecAdapter) NewNode(config *NodeConfig) (Node, error) { conf.Stack.P2P.ListenAddr = "127.0.0.1:0" node := &ExecNode{ - ID: config.ID, - Dir: dir, - Config: conf, + ID: config.ID, + Dir: dir, + Config: conf, + adapter: e, } node.newCmd = node.execCommand + e.nodes[node.ID] = node return node, nil } @@ -108,10 +115,11 @@ type ExecNode struct { Cmd *exec.Cmd Info *p2p.NodeInfo - client *rpc.Client - wsAddr string - newCmd func() *exec.Cmd - key *ecdsa.PrivateKey + adapter *ExecAdapter + client *rpc.Client + wsAddr string + newCmd func() *exec.Cmd + key *ecdsa.PrivateKey } // Addr returns the node's enode URL @@ -149,6 +157,10 @@ func (n *ExecNode) Start(snapshots map[string][]byte) (err error) { // encode a copy of the config containing the snapshot confCopy := *n.Config confCopy.Snapshots = snapshots + confCopy.PeerAddrs = make(map[string]string) + for id, node := range n.adapter.nodes { + confCopy.PeerAddrs[id.String()] = node.wsAddr + } confData, err := json.Marshal(confCopy) if err != nil { return fmt.Errorf("error generating node config: %s", err) @@ -315,7 +327,8 @@ func init() { type execNodeConfig struct { Stack node.Config `json:"stack"` Node *NodeConfig `json:"node"` - Snapshots map[string][]byte `json:"snapshot,omitempty"` + Snapshots map[string][]byte `json:"snapshots,omitempty"` + PeerAddrs map[string]string `json:"peer_addrs,omitempty"` } // execP2PNode starts a devp2p node when the current binary is executed with @@ -326,9 +339,8 @@ func execP2PNode() { glogger.Verbosity(log.LvlInfo) log.Root().SetHandler(glogger) - // read the services and ID from argv + // read the services from argv serviceNames := strings.Split(os.Args[1], ",") - id := discover.MustHexID(os.Args[2]) // decode the config confEnv := os.Getenv("_P2P_NODE_CONFIG") @@ -341,20 +353,6 @@ func execP2PNode() { } conf.Stack.P2P.PrivateKey = conf.Node.PrivateKey - // initialize the services - services := make(map[string]node.Service, len(serviceNames)) - for _, name := range serviceNames { - serviceFunc, exists := serviceFuncs[name] - if !exists { - log.Crit(fmt.Sprintf("unknown node service %q", name)) - } - var snapshot []byte - if conf.Snapshots != nil { - snapshot = conf.Snapshots[name] - } - services[name] = serviceFunc(id, snapshot) - } - // use explicit IP address in ListenAddr so that Enode URL is usable if strings.HasPrefix(conf.Stack.P2P.ListenAddr, ":") { addrs, err := net.InterfaceAddrs() @@ -369,12 +367,54 @@ func execP2PNode() { } } - // start the devp2p stack - stack, err := startP2PNode(&conf.Stack, services) + // initialize the devp2p stack + stack, err := node.New(&conf.Stack) if err != nil { - log.Crit("error starting p2p node", "err", err) + log.Crit("error creating node stack", "err", err) } + // register the services, collecting them into a map so we can wrap + // them in a snapshot service + services := make(map[string]node.Service, len(serviceNames)) + for _, name := range serviceNames { + serviceFunc, exists := serviceFuncs[name] + if !exists { + log.Crit("unknown node service", "name", name) + } + constructor := func(nodeCtx *node.ServiceContext) (node.Service, error) { + ctx := &ServiceContext{ + RPCDialer: &wsRPCDialer{addrs: conf.PeerAddrs}, + NodeContext: nodeCtx, + Config: conf.Node, + } + if conf.Snapshots != nil { + ctx.Snapshot = conf.Snapshots[name] + } + service, err := serviceFunc(ctx) + if err != nil { + return nil, err + } + services[name] = service + return service, nil + } + if err := stack.Register(constructor); err != nil { + log.Crit("error starting service", "name", name, "err", err) + } + } + + // register the snapshot service + if err := stack.Register(func(ctx *node.ServiceContext) (node.Service, error) { + return &snapshotService{services}, nil + }); err != nil { + log.Crit("error starting snapshot service", "err", err) + } + + // start the stack + if err := stack.Start(); err != nil { + log.Crit("error stating node stack", "err", err) + } + + // stop the stack if we get a SIGTERM signal go func() { sigc := make(chan os.Signal, 1) signal.Notify(sigc, syscall.SIGTERM) @@ -384,33 +424,10 @@ func execP2PNode() { stack.Stop() }() + // wait for the stack to exit stack.Wait() } -func startP2PNode(conf *node.Config, services map[string]node.Service) (*node.Node, error) { - stack, err := node.New(conf) - if err != nil { - return nil, err - } - constructor := func(service node.Service) func(ctx *node.ServiceContext) (node.Service, error) { - return func(ctx *node.ServiceContext) (node.Service, error) { - return service, nil - } - } - for _, service := range services { - if err := stack.Register(constructor(service)); err != nil { - return nil, err - } - } - if err := stack.Register(constructor(&snapshotService{services})); err != nil { - return nil, err - } - if err := stack.Start(); err != nil { - return nil, err - } - return stack, nil -} - // snapshotService is a node.Service which wraps a list of services and // exposes an API to generate a snapshot of those services type snapshotService struct { @@ -457,3 +474,15 @@ func (api SnapshotAPI) Snapshot() (map[string][]byte, error) { } return snapshots, nil } + +type wsRPCDialer struct { + addrs map[string]string +} + +func (w *wsRPCDialer) DialRPC(id discover.NodeID) (*rpc.Client, error) { + addr, ok := w.addrs[id.String()] + if !ok { + return nil, fmt.Errorf("unknown node: %s", id) + } + return rpc.DialWebsocket(context.Background(), addr, "http://localhost") +} diff --git a/p2p/simulations/adapters/inproc.go b/p2p/simulations/adapters/inproc.go index f0a989d829..cf3a404d73 100644 --- a/p2p/simulations/adapters/inproc.go +++ b/p2p/simulations/adapters/inproc.go @@ -97,6 +97,24 @@ func (s *SimAdapter) Dial(dest *discover.Node) (conn net.Conn, err error) { return pipe2, nil } +func (s *SimAdapter) DialRPC(id discover.NodeID) (*rpc.Client, error) { + simNode, ok := s.GetNode(id) + if !ok { + return nil, fmt.Errorf("unknown node: %s", id) + } + simNode.lock.RLock() + node := simNode.node + simNode.lock.RUnlock() + if node == nil { + return nil, errors.New("node not started") + } + handler, err := node.RPCHandler() + if err != nil { + return nil, err + } + return rpc.DialInProc(handler), nil +} + // GetNode returns the node with the given ID if it exists func (s *SimAdapter) GetNode(id discover.NodeID) (*SimNode, bool) { s.mtx.RLock() @@ -180,13 +198,20 @@ func (self *SimNode) Start(snapshots map[string][]byte) error { } newService := func(name string) func(ctx *node.ServiceContext) (node.Service, error) { - return func(ctx *node.ServiceContext) (node.Service, error) { - var snapshot []byte + return func(nodeCtx *node.ServiceContext) (node.Service, error) { + ctx := &ServiceContext{ + RPCDialer: self.adapter, + NodeContext: nodeCtx, + Config: self.config, + } if snapshots != nil { - snapshot = snapshots[name] + ctx.Snapshot = snapshots[name] } serviceFunc := self.adapter.services[name] - service := serviceFunc(self.ID, snapshot) + service, err := serviceFunc(ctx) + if err != nil { + return nil, err + } self.running = append(self.running, service) return service, nil } diff --git a/p2p/simulations/adapters/types.go b/p2p/simulations/adapters/types.go index 034d162297..f5ac3ec4ac 100644 --- a/p2p/simulations/adapters/types.go +++ b/p2p/simulations/adapters/types.go @@ -158,11 +158,23 @@ func RandomNodeConfig() *NodeConfig { } } +type ServiceContext struct { + RPCDialer + + NodeContext *node.ServiceContext + Config *NodeConfig + Snapshot []byte +} + +type RPCDialer interface { + DialRPC(id discover.NodeID) (*rpc.Client, error) +} + // Services is a collection of services which can be run in a simulation type Services map[string]ServiceFunc // ServiceFunc returns a node.Service which can be used to boot devp2p nodes -type ServiceFunc func(id discover.NodeID, snapshot []byte) node.Service +type ServiceFunc func(ctx *ServiceContext) (node.Service, error) // serviceFuncs is a map of registered services which are used to boot devp2p // nodes diff --git a/p2p/simulations/examples/connectivity.go b/p2p/simulations/examples/connectivity.go index 6174c094e8..95b7c8be7e 100644 --- a/p2p/simulations/examples/connectivity.go +++ b/p2p/simulations/examples/connectivity.go @@ -27,8 +27,8 @@ func main() { log.Root().SetHandler(log.LvlFilterHandler(log.LvlTrace, log.StreamHandler(os.Stderr, log.TerminalFormat(false)))) services := map[string]adapters.ServiceFunc{ - "ping-pong": func(id discover.NodeID, snapshot []byte) node.Service { - return newPingPongService(id) + "ping-pong": func(ctx *adapters.ServiceContext) (node.Service, error) { + return newPingPongService(ctx.Config.ID), nil }, } diff --git a/p2p/simulations/http_test.go b/p2p/simulations/http_test.go index d91127450e..84ceb0f677 100644 --- a/p2p/simulations/http_test.go +++ b/p2p/simulations/http_test.go @@ -24,10 +24,10 @@ type testService struct { state atomic.Value } -func newTestService(id discover.NodeID, snapshot []byte) node.Service { - svc := &testService{id: id} - svc.state.Store(snapshot) - return svc +func newTestService(ctx *adapters.ServiceContext) (node.Service, error) { + svc := &testService{id: ctx.Config.ID} + svc.state.Store(ctx.Snapshot) + return svc, nil } func (t *testService) Protocols() []p2p.Protocol { diff --git a/p2p/testing/protocoltester.go b/p2p/testing/protocoltester.go index bbaa18a8ae..9e58a79d18 100644 --- a/p2p/testing/protocoltester.go +++ b/p2p/testing/protocoltester.go @@ -21,11 +21,11 @@ type ProtocolTester struct { func NewProtocolTester(t *testing.T, id discover.NodeID, n int, run func(*p2p.Peer, p2p.MsgReadWriter) error) *ProtocolTester { services := adapters.Services{ - "test": func(id discover.NodeID, _ []byte) node.Service { - return &testNode{run} + "test": func(ctx *adapters.ServiceContext) (node.Service, error) { + return &testNode{run}, nil }, - "mock": func(id discover.NodeID, _ []byte) node.Service { - return newMockNode() + "mock": func(ctx *adapters.ServiceContext) (node.Service, error) { + return newMockNode(), nil }, } adapter := adapters.NewSimAdapter(services) diff --git a/swarm/network/simulations/discovery/discovery_test.go b/swarm/network/simulations/discovery/discovery_test.go index 13d1f1fbef..fb038ed2a7 100644 --- a/swarm/network/simulations/discovery/discovery_test.go +++ b/swarm/network/simulations/discovery/discovery_test.go @@ -179,8 +179,8 @@ func triggerChecks(trigger chan discover.NodeID, net *simulations.Network, id di return nil } -func newService(id discover.NodeID, snapshot []byte) node.Service { - addr := network.NewAddrFromNodeID(id) +func newService(ctx *adapters.ServiceContext) (node.Service, error) { + addr := network.NewAddrFromNodeID(ctx.Config.ID) kp := network.NewKadParams() kp.MinProxBinSize = testMinProxBinSize @@ -200,5 +200,5 @@ func newService(id discover.NodeID, snapshot []byte) node.Service { HiveParams: hp, } - return network.NewBzz(config, kad, nil) + return network.NewBzz(config, kad, nil), nil } diff --git a/swarm/network/simulations/overlay.go b/swarm/network/simulations/overlay.go index dbf3ddb867..319eeec5c4 100644 --- a/swarm/network/simulations/overlay.go +++ b/swarm/network/simulations/overlay.go @@ -33,7 +33,8 @@ func NewSimulation() *Simulation { } } -func (s *Simulation) NewService(id discover.NodeID, snapshot []byte) node.Service { +func (s *Simulation) NewService(ctx *adapters.ServiceContext) (node.Service, error) { + id := ctx.Config.ID s.mtx.Lock() store, ok := s.stores[id] if !ok { @@ -64,7 +65,7 @@ func (s *Simulation) NewService(id discover.NodeID, snapshot []byte) node.Servic HiveParams: hp, } - return network.NewBzz(config, kad, store) + return network.NewBzz(config, kad, store), nil } func createMockers() map[string]*simulations.MockerConfig { diff --git a/swarm/pss/pss_test.go b/swarm/pss/pss_test.go index f540d6aa3e..3b400ade20 100644 --- a/swarm/pss/pss_test.go +++ b/swarm/pss/pss_test.go @@ -453,16 +453,15 @@ func newServices() adapters.Services { return kademlias[id] } return adapters.Services{ - "pss": func(id discover.NodeID, snapshot []byte) node.Service { + "pss": func(ctx *adapters.ServiceContext) (node.Service, error) { + id := ctx.Config.ID cachedir, err := ioutil.TempDir("", "pss-cache") if err != nil { - log.Error("create pss cache tmpdir failed", "error", err) - return nil + return nil, err } dpa, err := storage.NewLocalDPA(cachedir) if err != nil { - log.Error("local dpa creation failed", "error", err) - return nil + return nil, err } pssp := NewPssParams() @@ -473,20 +472,20 @@ func newServices() adapters.Services { } err = RegisterPssProtocol(ps, &pssPingTopic, pssPingProtocol, newPssPingProtocol(ping.pssPingHandler)) if err != nil { - log.Error("Couldnt register pss protocol", "err", err) - os.Exit(1) + return nil, err } - return ps + return ps, nil }, - "bzz": func(id discover.NodeID, snapshot []byte) node.Service { + "bzz": func(ctx *adapters.ServiceContext) (node.Service, error) { + id := ctx.Config.ID addr := network.NewAddrFromNodeID(id) config := &network.BzzConfig{ OverlayAddr: addr.Over(), UnderlayAddr: addr.Under(), HiveParams: network.NewHiveParams(), } - return network.NewBzz(config, kademlia(id), stateStore) + return network.NewBzz(config, kademlia(id), stateStore), nil }, } } From a92a34145016dc7e4030290e98ef8f414ffe7e97 Mon Sep 17 00:00:00 2001 From: Lewis Marshall Date: Mon, 5 Jun 2017 10:14:00 +0100 Subject: [PATCH 4/5] p2p/simulations: Add external networks to HTTP server Signed-off-by: Lewis Marshall --- p2p/simulations/http.go | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/p2p/simulations/http.go b/p2p/simulations/http.go index 1b86fca438..0d1742cfaf 100644 --- a/p2p/simulations/http.go +++ b/p2p/simulations/http.go @@ -255,6 +255,10 @@ type ServerConfig struct { // network NewAdapter func() adapters.NodeAdapter + // ExternalNetworks are externally defined networks to expose via the + // HTTP server + ExternalNetworks map[string]*Network + // Mocker is the function which will be called when a client sends a // POST request to /networks//mock and is expected to // generate some mock events in the network @@ -286,6 +290,9 @@ func NewServer(config *ServerConfig) *Server { router: httprouter.New(), networks: make(map[string]*Network), } + for name, network := range config.ExternalNetworks { + s.networks[name] = network + } s.OPTIONS("/networks", s.Options) s.POST("/networks", s.CreateNetwork) From 479409d1f3d077618bcdbea3e94b9bb463b85fd0 Mon Sep 17 00:00:00 2001 From: Lewis Marshall Date: Mon, 5 Jun 2017 10:51:43 +0100 Subject: [PATCH 5/5] p2p/simulations: Fix creating snapshots Signed-off-by: Lewis Marshall --- p2p/simulations/adapters/inproc.go | 33 ++++++++++++++++++++++-------- 1 file changed, 25 insertions(+), 8 deletions(-) diff --git a/p2p/simulations/adapters/inproc.go b/p2p/simulations/adapters/inproc.go index cf3a404d73..f81611a470 100644 --- a/p2p/simulations/adapters/inproc.go +++ b/p2p/simulations/adapters/inproc.go @@ -78,6 +78,7 @@ func (s *SimAdapter) NewNode(config *NodeConfig) (Node, error) { ID: id, config: config, adapter: s, + running: make(map[string]node.Service), } s.nodes[id] = node return node, nil @@ -135,7 +136,7 @@ type SimNode struct { config *NodeConfig adapter *SimAdapter node *node.Node - running []node.Service + running map[string]node.Service client *rpc.Client } @@ -181,12 +182,24 @@ func (self *SimNode) ServeRPC(conn net.Conn) error { // simulation_snapshot RPC method func (self *SimNode) Snapshots() (map[string][]byte, error) { self.lock.Lock() - defer self.lock.Unlock() - if self.client == nil { - return nil, errors.New("RPC not started") + services := self.running + self.lock.Unlock() + if len(services) == 0 { + return nil, errors.New("no running services") } - var snapshots map[string][]byte - return snapshots, self.client.Call(&snapshots, "simulation_snapshot") + snapshots := make(map[string][]byte) + for name, service := range services { + if s, ok := service.(interface { + Snapshot() ([]byte, error) + }); ok { + snap, err := s.Snapshot() + if err != nil { + return nil, err + } + snapshots[name] = snap + } + } + return snapshots, nil } // Start starts the RPC handler and the underlying service @@ -212,7 +225,7 @@ func (self *SimNode) Start(snapshots map[string][]byte) error { if err != nil { return nil, err } - self.running = append(self.running, service) + self.running[name] = service return service, nil } } @@ -270,7 +283,11 @@ func (self *SimNode) Stop() error { func (self *SimNode) Services() []node.Service { self.lock.Lock() defer self.lock.Unlock() - return self.running + services := make([]node.Service, 0, len(self.running)) + for _, service := range self.running { + services = append(services, service) + } + return services } func (self *SimNode) Server() *p2p.Server {