plugeth/swarm/network/protocol.go

// 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 <http://www.gnu.org/licenses/>.

package network

/*
bzz implements the swarm wire protocol [bzz] (sister of eth and shh)
the protocol instance is launched on each peer by the network layer if the
bzz protocol handler is registered on the p2p server.

The bzz protocol component speaks the bzz protocol
* handle the protocol handshake
* register peers in the KΛÐΞMLIΛ table via the hive logistic manager
* dispatch to hive for handling the DHT logic
* encode and decode requests for storage and retrieval
* handle sync protocol messages via the syncer
* talks the SWAP payment protocol (swap accounting is done within NetStore)
*/

import (
	"errors"
	"fmt"
	"net"
	"strconv"
	"time"

	"github.com/ethereum/go-ethereum/contracts/chequebook"
	"github.com/ethereum/go-ethereum/log"
	"github.com/ethereum/go-ethereum/metrics"
	"github.com/ethereum/go-ethereum/p2p"
	bzzswap "github.com/ethereum/go-ethereum/swarm/services/swap"
	"github.com/ethereum/go-ethereum/swarm/services/swap/swap"
	"github.com/ethereum/go-ethereum/swarm/storage"
)

//metrics variables
var (
	storeRequestMsgCounter    = metrics.NewRegisteredCounter("network.protocol.msg.storerequest.count", nil)
	retrieveRequestMsgCounter = metrics.NewRegisteredCounter("network.protocol.msg.retrieverequest.count", nil)
	peersMsgCounter           = metrics.NewRegisteredCounter("network.protocol.msg.peers.count", nil)
	syncRequestMsgCounter     = metrics.NewRegisteredCounter("network.protocol.msg.syncrequest.count", nil)
	unsyncedKeysMsgCounter    = metrics.NewRegisteredCounter("network.protocol.msg.unsyncedkeys.count", nil)
	deliverRequestMsgCounter  = metrics.NewRegisteredCounter("network.protocol.msg.deliverrequest.count", nil)
	paymentMsgCounter         = metrics.NewRegisteredCounter("network.protocol.msg.payment.count", nil)
	invalidMsgCounter         = metrics.NewRegisteredCounter("network.protocol.msg.invalid.count", nil)
	handleStatusMsgCounter    = metrics.NewRegisteredCounter("network.protocol.msg.handlestatus.count", nil)
)

const (
	Version            = 0
	ProtocolLength     = uint64(8)
	ProtocolMaxMsgSize = 10 * 1024 * 1024
	NetworkId          = 3
)

// bzz represents the swarm wire protocol
// an instance is running on each peer
type bzz struct {
	storage    StorageHandler       // handler storage/retrieval related requests coming via the bzz wire protocol
	hive       *Hive                // the logistic manager, peerPool, routing service and peer handler
	dbAccess   *DbAccess            // access to db storage counter and iterator for syncing
	requestDb  *storage.LDBDatabase // db to persist backlog of deliveries to aid syncing
	remoteAddr *peerAddr            // remote peers address
	peer       *p2p.Peer            // the p2p peer object
	rw         p2p.MsgReadWriter    // messageReadWriter to send messages to
	backend    chequebook.Backend
	lastActive time.Time
	NetworkId  uint64

	swap        *swap.Swap          // swap instance for the peer connection
	swapParams  *bzzswap.SwapParams // swap settings both local and remote
	swapEnabled bool                // flag to enable SWAP (will be set via Caps in handshake)
	syncEnabled bool                // flag to enable SYNC (will be set via Caps in handshake)
	syncer      *syncer             // syncer instance for the peer connection
	syncParams  *SyncParams         // syncer params
	syncState   *syncState          // outgoing syncronisation state (contains reference to remote peers db counter)
}

// interface type for handler of storage/retrieval related requests coming
// via the bzz wire protocol
// messages: UnsyncedKeys, DeliveryRequest, StoreRequest, RetrieveRequest
type StorageHandler interface {
	HandleUnsyncedKeysMsg(req *unsyncedKeysMsgData, p *peer) error
	HandleDeliveryRequestMsg(req *deliveryRequestMsgData, p *peer) error
	HandleStoreRequestMsg(req *storeRequestMsgData, p *peer)
	HandleRetrieveRequestMsg(req *retrieveRequestMsgData, p *peer)
}

/*
main entrypoint, wrappers starting a server that will run the bzz protocol
use this constructor to attach the protocol ("class") to server caps
This is done by node.Node#Register(func(node.ServiceContext) (Service, error))
Service implements Protocols() which is an array of protocol constructors
at node startup the protocols are initialised
the Dev p2p layer then calls Run(p *p2p.Peer, rw p2p.MsgReadWriter) error
on each peer connection
The Run function of the Bzz protocol class creates a bzz instance
which will represent the peer for the swarm hive and all peer-aware components
*/
func Bzz(cloud StorageHandler, backend chequebook.Backend, hive *Hive, dbaccess *DbAccess, sp *bzzswap.SwapParams, sy *SyncParams, networkId uint64) (p2p.Protocol, error) {

	// a single global request db is created for all peer connections
	// this is to persist delivery backlog and aid syncronisation
	requestDb, err := storage.NewLDBDatabase(sy.RequestDbPath)
	if err != nil {
		return p2p.Protocol{}, fmt.Errorf("error setting up request db: %v", err)
	}
	if networkId == 0 {
		networkId = NetworkId
	}
	return p2p.Protocol{
		Name:    "bzz",
		Version: Version,
		Length:  ProtocolLength,
		Run: func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
			return run(requestDb, cloud, backend, hive, dbaccess, sp, sy, networkId, p, rw)
		},
	}, nil
}

/*
the main protocol loop that
 * does the handshake by exchanging statusMsg
 * if peer is valid and accepted, registers with the hive
 * then enters into a forever loop handling incoming messages
 * storage and retrieval related queries coming via bzz are dispatched to StorageHandler
 * peer-related messages are dispatched to the hive
 * payment related messages are relayed to SWAP service
 * on disconnect, unregister the peer in the hive (note RemovePeer in the post-disconnect hook)
 * whenever the loop terminates, the peer will disconnect with Subprotocol error
 * whenever handlers return an error the loop terminates
*/
func run(requestDb *storage.LDBDatabase, depo StorageHandler, backend chequebook.Backend, hive *Hive, dbaccess *DbAccess, sp *bzzswap.SwapParams, sy *SyncParams, networkId uint64, p *p2p.Peer, rw p2p.MsgReadWriter) (err error) {

	self := &bzz{
		storage:     depo,
		backend:     backend,
		hive:        hive,
		dbAccess:    dbaccess,
		requestDb:   requestDb,
		peer:        p,
		rw:          rw,
		swapParams:  sp,
		syncParams:  sy,
		swapEnabled: hive.swapEnabled,
		syncEnabled: true,
		NetworkId:   networkId,
	}

	// handle handshake
	err = self.handleStatus()
	if err != nil {
		return err
	}
	defer func() {
		// if the handler loop exits, the peer is disconnecting
		// deregister the peer in the hive
		self.hive.removePeer(&peer{bzz: self})
		if self.syncer != nil {
			self.syncer.stop() // quits request db and delivery loops, save requests
		}
		if self.swap != nil {
			self.swap.Stop() // quits chequebox autocash etc
		}
	}()

	// the main forever loop that handles incoming requests
	for {
		if self.hive.blockRead {
			log.Warn(fmt.Sprintf("Cannot read network"))
			time.Sleep(100 * time.Millisecond)
			continue
		}
		err = self.handle()
		if err != nil {
			return
		}
	}
}

// TODO: may need to implement protocol drop only? don't want to kick off the peer
// if they are useful for other protocols
func (self *bzz) Drop() {
	self.peer.Disconnect(p2p.DiscSubprotocolError)
}

// one cycle of the main forever loop that handles and dispatches incoming messages
func (self *bzz) handle() error {
	msg, err := self.rw.ReadMsg()
	log.Debug(fmt.Sprintf("<- %v", msg))
	if err != nil {
		return err
	}
	if msg.Size > ProtocolMaxMsgSize {
		return fmt.Errorf("message too long: %v > %v", msg.Size, ProtocolMaxMsgSize)
	}
	// make sure that the payload has been fully consumed
	defer msg.Discard()

	switch msg.Code {

	case statusMsg:
		// no extra status message allowed. The one needed already handled by
		// handleStatus
		log.Debug(fmt.Sprintf("Status message: %v", msg))
		return errors.New("extra status message")

	case storeRequestMsg:
		// store requests are dispatched to netStore
		storeRequestMsgCounter.Inc(1)
		var req storeRequestMsgData
		if err := msg.Decode(&req); err != nil {
			return fmt.Errorf("<- %v: %v", msg, err)
		}
		if n := len(req.SData); n < 9 {
			return fmt.Errorf("<- %v: Data too short (%v)", msg, n)
		}
		// last Active time is set only when receiving chunks
		self.lastActive = time.Now()
		log.Trace(fmt.Sprintf("incoming store request: %s", req.String()))
		// swap accounting is done within forwarding
		self.storage.HandleStoreRequestMsg(&req, &peer{bzz: self})

	case retrieveRequestMsg:
		// retrieve Requests are dispatched to netStore
		retrieveRequestMsgCounter.Inc(1)
		var req retrieveRequestMsgData
		if err := msg.Decode(&req); err != nil {
			return fmt.Errorf("<- %v: %v", msg, err)
		}
		req.from = &peer{bzz: self}
		// if request is lookup and not to be delivered
		if req.isLookup() {
			log.Trace(fmt.Sprintf("self lookup for %v: responding with peers only...", req.from))
		} else if req.Key == nil {
			return fmt.Errorf("protocol handler: req.Key == nil || req.Timeout == nil")
		} else {
			// swap accounting is done within netStore
			self.storage.HandleRetrieveRequestMsg(&req, &peer{bzz: self})
		}
		// direct response with peers, TODO: sort this out
		self.hive.peers(&req)

	case peersMsg:
		// response to lookups and immediate response to retrieve requests
		// dispatches new peer data to the hive that adds them to KADDB
		peersMsgCounter.Inc(1)
		var req peersMsgData
		if err := msg.Decode(&req); err != nil {
			return fmt.Errorf("<- %v: %v", msg, err)
		}
		req.from = &peer{bzz: self}
		log.Trace(fmt.Sprintf("<- peer addresses: %v", req))
		self.hive.HandlePeersMsg(&req, &peer{bzz: self})

	case syncRequestMsg:
		syncRequestMsgCounter.Inc(1)
		var req syncRequestMsgData
		if err := msg.Decode(&req); err != nil {
			return fmt.Errorf("<- %v: %v", msg, err)
		}
		log.Debug(fmt.Sprintf("<- sync request: %v", req))
		self.lastActive = time.Now()
		self.sync(req.SyncState)

	case unsyncedKeysMsg:
		// coming from parent node offering
		unsyncedKeysMsgCounter.Inc(1)
		var req unsyncedKeysMsgData
		if err := msg.Decode(&req); err != nil {
			return fmt.Errorf("<- %v: %v", msg, err)
		}
		log.Debug(fmt.Sprintf("<- unsynced keys : %s", req.String()))
		err := self.storage.HandleUnsyncedKeysMsg(&req, &peer{bzz: self})
		self.lastActive = time.Now()
		if err != nil {
			return fmt.Errorf("<- %v: %v", msg, err)
		}

	case deliveryRequestMsg:
		// response to syncKeysMsg hashes filtered not existing in db
		// also relays the last synced state to the source
		deliverRequestMsgCounter.Inc(1)
		var req deliveryRequestMsgData
		if err := msg.Decode(&req); err != nil {
			return fmt.Errorf("<-msg %v: %v", msg, err)
		}
		log.Debug(fmt.Sprintf("<- delivery request: %s", req.String()))
		err := self.storage.HandleDeliveryRequestMsg(&req, &peer{bzz: self})
		self.lastActive = time.Now()
		if err != nil {
			return fmt.Errorf("<- %v: %v", msg, err)
		}

	case paymentMsg:
		// swap protocol message for payment, Units paid for, Cheque paid with
		paymentMsgCounter.Inc(1)
		if self.swapEnabled {
			var req paymentMsgData
			if err := msg.Decode(&req); err != nil {
				return fmt.Errorf("<- %v: %v", msg, err)
			}
			log.Debug(fmt.Sprintf("<- payment: %s", req.String()))
			self.swap.Receive(int(req.Units), req.Promise)
		}

	default:
		// no other message is allowed
		invalidMsgCounter.Inc(1)
		return fmt.Errorf("invalid message code: %v", msg.Code)
	}
	return nil
}

func (self *bzz) handleStatus() (err error) {

	handshake := &statusMsgData{
		Version:   uint64(Version),
		ID:        "honey",
		Addr:      self.selfAddr(),
		NetworkId: self.NetworkId,
		Swap: &bzzswap.SwapProfile{
			Profile:    self.swapParams.Profile,
			PayProfile: self.swapParams.PayProfile,
		},
	}

	err = p2p.Send(self.rw, statusMsg, handshake)
	if err != nil {
		return err
	}

	// read and handle remote status
	var msg p2p.Msg
	msg, err = self.rw.ReadMsg()
	if err != nil {
		return err
	}

	if msg.Code != statusMsg {
		return fmt.Errorf("first msg has code %x (!= %x)", msg.Code, statusMsg)
	}

	handleStatusMsgCounter.Inc(1)

	if msg.Size > ProtocolMaxMsgSize {
		return fmt.Errorf("message too long: %v > %v", msg.Size, ProtocolMaxMsgSize)
	}

	var status statusMsgData
	if err := msg.Decode(&status); err != nil {
		return fmt.Errorf("<- %v: %v", msg, err)
	}

	if status.NetworkId != self.NetworkId {
		return fmt.Errorf("network id mismatch: %d (!= %d)", status.NetworkId, self.NetworkId)
	}

	if Version != status.Version {
		return fmt.Errorf("protocol version mismatch: %d (!= %d)", status.Version, Version)
	}

	self.remoteAddr = self.peerAddr(status.Addr)
	log.Trace(fmt.Sprintf("self: advertised IP: %v, peer advertised: %v, local address: %v\npeer: advertised IP: %v, remote address: %v\n", self.selfAddr(), self.remoteAddr, self.peer.LocalAddr(), status.Addr.IP, self.peer.RemoteAddr()))

	if self.swapEnabled {
		// set remote profile for accounting
		self.swap, err = bzzswap.NewSwap(self.swapParams, status.Swap, self.backend, self)
		if err != nil {
			return err
		}
	}

	log.Info(fmt.Sprintf("Peer %08x is capable (%d/%d)", self.remoteAddr.Addr[:4], status.Version, status.NetworkId))
	err = self.hive.addPeer(&peer{bzz: self})
	if err != nil {
		return err
	}

	// hive sets syncstate so sync should start after node added
	log.Info(fmt.Sprintf("syncronisation request sent with %v", self.syncState))
	self.syncRequest()

	return nil
}

func (self *bzz) sync(state *syncState) error {
	// syncer setup
	if self.syncer != nil {
		return errors.New("sync request can only be sent once")
	}

	cnt := self.dbAccess.counter()
	remoteaddr := self.remoteAddr.Addr
	start, stop := self.hive.kad.KeyRange(remoteaddr)

	// an explicitly received nil syncstate disables syncronisation
	if state == nil {
		self.syncEnabled = false
		log.Warn(fmt.Sprintf("syncronisation disabled for peer %v", self))
		state = &syncState{DbSyncState: &storage.DbSyncState{}, Synced: true}
	} else {
		state.synced = make(chan bool)
		state.SessionAt = cnt
		if storage.IsZeroKey(state.Stop) && state.Synced {
			state.Start = storage.Key(start[:])
			state.Stop = storage.Key(stop[:])
		}
		log.Debug(fmt.Sprintf("syncronisation requested by peer %v at state %v", self, state))
	}
	var err error
	self.syncer, err = newSyncer(
		self.requestDb,
		storage.Key(remoteaddr[:]),
		self.dbAccess,
		self.unsyncedKeys, self.store,
		self.syncParams, state, func() bool { return self.syncEnabled },
	)
	if err != nil {
		return nil
	}
	log.Trace(fmt.Sprintf("syncer set for peer %v", self))
	return nil
}

func (self *bzz) String() string {
	return self.remoteAddr.String()
}

// repair reported address if IP missing
func (self *bzz) peerAddr(base *peerAddr) *peerAddr {
	if base.IP.IsUnspecified() {
		host, _, _ := net.SplitHostPort(self.peer.RemoteAddr().String())
		base.IP = net.ParseIP(host)
	}
	return base
}

// returns self advertised node connection info (listening address w enodes)
// IP will get repaired on the other end if missing
// or resolved via ID by discovery at dialout
func (self *bzz) selfAddr() *peerAddr {
	id := self.hive.id
	host, port, _ := net.SplitHostPort(self.hive.listenAddr())
	intport, _ := strconv.Atoi(port)
	addr := &peerAddr{
		Addr: self.hive.addr,
		ID:   id[:],
		IP:   net.ParseIP(host),
		Port: uint16(intport),
	}
	return addr
}

// outgoing messages
// send retrieveRequestMsg
func (self *bzz) retrieve(req *retrieveRequestMsgData) error {
	return self.send(retrieveRequestMsg, req)
}

// send storeRequestMsg
func (self *bzz) store(req *storeRequestMsgData) error {
	return self.send(storeRequestMsg, req)
}

func (self *bzz) syncRequest() error {
	req := &syncRequestMsgData{}
	if self.hive.syncEnabled {
		log.Debug(fmt.Sprintf("syncronisation request to peer %v at state %v", self, self.syncState))
		req.SyncState = self.syncState
	}
	if self.syncState == nil {
		log.Warn(fmt.Sprintf("syncronisation disabled for peer %v at state %v", self, self.syncState))
	}
	return self.send(syncRequestMsg, req)
}

// queue storeRequestMsg in request db
func (self *bzz) deliveryRequest(reqs []*syncRequest) error {
	req := &deliveryRequestMsgData{
		Deliver: reqs,
	}
	return self.send(deliveryRequestMsg, req)
}

// batch of syncRequests to send off
func (self *bzz) unsyncedKeys(reqs []*syncRequest, state *syncState) error {
	req := &unsyncedKeysMsgData{
		Unsynced: reqs,
		State:    state,
	}
	return self.send(unsyncedKeysMsg, req)
}

// send paymentMsg
func (self *bzz) Pay(units int, promise swap.Promise) {
	req := &paymentMsgData{uint(units), promise.(*chequebook.Cheque)}
	self.payment(req)
}

// send paymentMsg
func (self *bzz) payment(req *paymentMsgData) error {
	return self.send(paymentMsg, req)
}

// sends peersMsg
func (self *bzz) peers(req *peersMsgData) error {
	return self.send(peersMsg, req)
}

func (self *bzz) send(msg uint64, data interface{}) error {
	if self.hive.blockWrite {
		return fmt.Errorf("network write blocked")
	}
	log.Trace(fmt.Sprintf("-> %v: %v (%T) to %v", msg, data, data, self))
	err := p2p.Send(self.rw, msg, data)
	if err != nil {
		self.Drop()
	}
	return err
}