swarm: network rewrite merge

1 files changed, 354 insertions, 0 deletions

diff --git a/swarm/pss/client/client.go b/swarm/pss/client/client.go
new file mode 100644
index 000000000..532a22384
--- /dev/null
+++ b/swarm/pss/client/client.go
@@ -0,0 +1,354 @@
+// Copyright 2018 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 <http://www.gnu.org/licenses/>.
+
+// +build !noclient,!noprotocol
+
+package client
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/ethereum/go-ethereum/common/hexutil"
+	"github.com/ethereum/go-ethereum/p2p"
+	"github.com/ethereum/go-ethereum/p2p/discover"
+	"github.com/ethereum/go-ethereum/p2p/protocols"
+	"github.com/ethereum/go-ethereum/rlp"
+	"github.com/ethereum/go-ethereum/rpc"
+	"github.com/ethereum/go-ethereum/swarm/log"
+	"github.com/ethereum/go-ethereum/swarm/pss"
+)
+
+const (
+	handshakeRetryTimeout = 1000
+	handshakeRetryCount   = 3
+)
+
+// The pss client provides devp2p emulation over pss RPC API,
+// giving access to pss methods from a different process
+type Client struct {
+	BaseAddrHex string
+
+	// peers
+	peerPool map[pss.Topic]map[string]*pssRPCRW
+	protos   map[pss.Topic]*p2p.Protocol
+
+	// rpc connections
+	rpc  *rpc.Client
+	subs []*rpc.ClientSubscription
+
+	// channels
+	topicsC chan []byte
+	quitC   chan struct{}
+
+	poolMu sync.Mutex
+}
+
+// implements p2p.MsgReadWriter
+type pssRPCRW struct {
+	*Client
+	topic    string
+	msgC     chan []byte
+	addr     pss.PssAddress
+	pubKeyId string
+	lastSeen time.Time
+	closed   bool
+}
+
+func (c *Client) newpssRPCRW(pubkeyid string, addr pss.PssAddress, topicobj pss.Topic) (*pssRPCRW, error) {
+	topic := topicobj.String()
+	err := c.rpc.Call(nil, "pss_setPeerPublicKey", pubkeyid, topic, hexutil.Encode(addr[:]))
+	if err != nil {
+		return nil, fmt.Errorf("setpeer %s %s: %v", topic, pubkeyid, err)
+	}
+	return &pssRPCRW{
+		Client:   c,
+		topic:    topic,
+		msgC:     make(chan []byte),
+		addr:     addr,
+		pubKeyId: pubkeyid,
+	}, nil
+}
+
+func (rw *pssRPCRW) ReadMsg() (p2p.Msg, error) {
+	msg := <-rw.msgC
+	log.Trace("pssrpcrw read", "msg", msg)
+	pmsg, err := pss.ToP2pMsg(msg)
+	if err != nil {
+		return p2p.Msg{}, err
+	}
+
+	return pmsg, nil
+}
+
+// If only one message slot left
+// then new is requested through handshake
+// if buffer is empty, handshake request blocks until return
+// after which pointer is changed to first new key in buffer
+// will fail if:
+// - any api calls fail
+// - handshake retries are exhausted without reply,
+// - send fails
+func (rw *pssRPCRW) WriteMsg(msg p2p.Msg) error {
+	log.Trace("got writemsg pssclient", "msg", msg)
+	if rw.closed {
+		return fmt.Errorf("connection closed")
+	}
+	rlpdata := make([]byte, msg.Size)
+	msg.Payload.Read(rlpdata)
+	pmsg, err := rlp.EncodeToBytes(pss.ProtocolMsg{
+		Code:    msg.Code,
+		Size:    msg.Size,
+		Payload: rlpdata,
+	})
+	if err != nil {
+		return err
+	}
+
+	// Get the keys we have
+	var symkeyids []string
+	err = rw.Client.rpc.Call(&symkeyids, "pss_getHandshakeKeys", rw.pubKeyId, rw.topic, false, true)
+	if err != nil {
+		return err
+	}
+
+	// Check the capacity of the first key
+	var symkeycap uint16
+	if len(symkeyids) > 0 {
+		err = rw.Client.rpc.Call(&symkeycap, "pss_getHandshakeKeyCapacity", symkeyids[0])
+		if err != nil {
+			return err
+		}
+	}
+
+	err = rw.Client.rpc.Call(nil, "pss_sendSym", symkeyids[0], rw.topic, hexutil.Encode(pmsg))
+	if err != nil {
+		return err
+	}
+
+	// If this is the last message it is valid for, initiate new handshake
+	if symkeycap == 1 {
+		var retries int
+		var sync bool
+		// if it's the only remaining key, make sure we don't continue until we have new ones for further writes
+		if len(symkeyids) == 1 {
+			sync = true
+		}
+		// initiate handshake
+		_, err := rw.handshake(retries, sync, false)
+		if err != nil {
+			log.Warn("failing", "err", err)
+			return err
+		}
+	}
+	return nil
+}
+
+// retry and synchronicity wrapper for handshake api call
+// returns first new symkeyid upon successful execution
+func (rw *pssRPCRW) handshake(retries int, sync bool, flush bool) (string, error) {
+
+	var symkeyids []string
+	var i int
+	// request new keys
+	// if the key buffer was depleted, make this as a blocking call and try several times before giving up
+	for i = 0; i < 1+retries; i++ {
+		log.Debug("handshake attempt pssrpcrw", "pubkeyid", rw.pubKeyId, "topic", rw.topic, "sync", sync)
+		err := rw.Client.rpc.Call(&symkeyids, "pss_handshake", rw.pubKeyId, rw.topic, sync, flush)
+		if err == nil {
+			var keyid string
+			if sync {
+				keyid = symkeyids[0]
+			}
+			return keyid, nil
+		}
+		if i-1+retries > 1 {
+			time.Sleep(time.Millisecond * handshakeRetryTimeout)
+		}
+	}
+
+	return "", fmt.Errorf("handshake failed after %d attempts", i)
+}
+
+// Custom constructor
+//
+// Provides direct access to the rpc object
+func NewClient(rpcurl string) (*Client, error) {
+	rpcclient, err := rpc.Dial(rpcurl)
+	if err != nil {
+		return nil, err
+	}
+
+	client, err := NewClientWithRPC(rpcclient)
+	if err != nil {
+		return nil, err
+	}
+	return client, nil
+}
+
+// Main constructor
+//
+// The 'rpcclient' parameter allows passing a in-memory rpc client to act as the remote websocket RPC.
+func NewClientWithRPC(rpcclient *rpc.Client) (*Client, error) {
+	client := newClient()
+	client.rpc = rpcclient
+	err := client.rpc.Call(&client.BaseAddrHex, "pss_baseAddr")
+	if err != nil {
+		return nil, fmt.Errorf("cannot get pss node baseaddress: %v", err)
+	}
+	return client, nil
+}
+
+func newClient() (client *Client) {
+	client = &Client{
+		quitC:    make(chan struct{}),
+		peerPool: make(map[pss.Topic]map[string]*pssRPCRW),
+		protos:   make(map[pss.Topic]*p2p.Protocol),
+	}
+	return
+}
+
+// Mounts a new devp2p protcool on the pss connection
+//
+// the protocol is aliased as a "pss topic"
+// uses normal devp2p send and incoming message handler routines from the p2p/protocols package
+//
+// when an incoming message is received from a peer that is not yet known to the client,
+// this peer object is instantiated, and the protocol is run on it.
+func (c *Client) RunProtocol(ctx context.Context, proto *p2p.Protocol) error {
+	topicobj := pss.BytesToTopic([]byte(fmt.Sprintf("%s:%d", proto.Name, proto.Version)))
+	topichex := topicobj.String()
+	msgC := make(chan pss.APIMsg)
+	c.peerPool[topicobj] = make(map[string]*pssRPCRW)
+	sub, err := c.rpc.Subscribe(ctx, "pss", msgC, "receive", topichex)
+	if err != nil {
+		return fmt.Errorf("pss event subscription failed: %v", err)
+	}
+	c.subs = append(c.subs, sub)
+	err = c.rpc.Call(nil, "pss_addHandshake", topichex)
+	if err != nil {
+		return fmt.Errorf("pss handshake activation failed: %v", err)
+	}
+
+	// dispatch incoming messages
+	go func() {
+		for {
+			select {
+			case msg := <-msgC:
+				// we only allow sym msgs here
+				if msg.Asymmetric {
+					continue
+				}
+				// we get passed the symkeyid
+				// need the symkey itself to resolve to peer's pubkey
+				var pubkeyid string
+				err = c.rpc.Call(&pubkeyid, "pss_getHandshakePublicKey", msg.Key)
+				if err != nil || pubkeyid == "" {
+					log.Trace("proto err or no pubkey", "err", err, "symkeyid", msg.Key)
+					continue
+				}
+				// if we don't have the peer on this protocol already, create it
+				// this is more or less the same as AddPssPeer, less the handshake initiation
+				if c.peerPool[topicobj][pubkeyid] == nil {
+					var addrhex string
+					err := c.rpc.Call(&addrhex, "pss_getAddress", topichex, false, msg.Key)
+					if err != nil {
+						log.Trace(err.Error())
+						continue
+					}
+					addrbytes, err := hexutil.Decode(addrhex)
+					if err != nil {
+						log.Trace(err.Error())
+						break
+					}
+					addr := pss.PssAddress(addrbytes)
+					rw, err := c.newpssRPCRW(pubkeyid, addr, topicobj)
+					if err != nil {
+						break
+					}
+					c.peerPool[topicobj][pubkeyid] = rw
+					nid, _ := discover.HexID("0x00")
+					p := p2p.NewPeer(nid, fmt.Sprintf("%v", addr), []p2p.Cap{})
+					go proto.Run(p, c.peerPool[topicobj][pubkeyid])
+				}
+				go func() {
+					c.peerPool[topicobj][pubkeyid].msgC <- msg.Msg
+				}()
+			case <-c.quitC:
+				return
+			}
+		}
+	}()
+
+	c.protos[topicobj] = proto
+	return nil
+}
+
+// Always call this to ensure that we exit cleanly
+func (c *Client) Close() error {
+	for _, s := range c.subs {
+		s.Unsubscribe()
+	}
+	return nil
+}
+
+// Add a pss peer (public key) and run the protocol on it
+//
+// client.RunProtocol with matching topic must have been
+// run prior to adding the peer, or this method will
+// return an error.
+//
+// The key must exist in the key store of the pss node
+// before the peer is added. The method will return an error
+// if it is not.
+func (c *Client) AddPssPeer(pubkeyid string, addr []byte, spec *protocols.Spec) error {
+	topic := pss.ProtocolTopic(spec)
+	if c.peerPool[topic] == nil {
+		return errors.New("addpeer on unset topic")
+	}
+	if c.peerPool[topic][pubkeyid] == nil {
+		rw, err := c.newpssRPCRW(pubkeyid, addr, topic)
+		if err != nil {
+			return err
+		}
+		_, err = rw.handshake(handshakeRetryCount, true, true)
+		if err != nil {
+			return err
+		}
+		c.poolMu.Lock()
+		c.peerPool[topic][pubkeyid] = rw
+		c.poolMu.Unlock()
+		nid, _ := discover.HexID("0x00")
+		p := p2p.NewPeer(nid, fmt.Sprintf("%v", addr), []p2p.Cap{})
+		go c.protos[topic].Run(p, c.peerPool[topic][pubkeyid])
+	}
+	return nil
+}
+
+// Remove a pss peer
+//
+// TODO: underlying cleanup
+func (c *Client) RemovePssPeer(pubkeyid string, spec *protocols.Spec) {
+	log.Debug("closing pss client peer", "pubkey", pubkeyid, "protoname", spec.Name, "protoversion", spec.Version)
+	c.poolMu.Lock()
+	defer c.poolMu.Unlock()
+	topic := pss.ProtocolTopic(spec)
+	c.peerPool[topic][pubkeyid].closed = true
+	delete(c.peerPool[topic], pubkeyid)
+}