diff options
Diffstat (limited to 'swarm/network/protocol.go')
| -rw-r--r-- | swarm/network/protocol.go | 759 |
1 files changed, 327 insertions, 432 deletions
diff --git a/swarm/network/protocol.go b/swarm/network/protocol.go index 1cbe00a97..39673f5a1 100644 --- a/swarm/network/protocol.go +++ b/swarm/network/protocol.go @@ -16,519 +16,414 @@ 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 ( + "context" "errors" "fmt" "net" - "strconv" + "sync" "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/crypto" "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) + "github.com/ethereum/go-ethereum/p2p/discover" + "github.com/ethereum/go-ethereum/p2p/protocols" + "github.com/ethereum/go-ethereum/rpc" + "github.com/ethereum/go-ethereum/swarm/log" + "github.com/ethereum/go-ethereum/swarm/state" ) const ( - Version = 0 - ProtocolLength = uint64(8) + DefaultNetworkID = 3 + // ProtocolMaxMsgSize maximum allowed message size ProtocolMaxMsgSize = 10 * 1024 * 1024 - NetworkId = 3 + // timeout for waiting + bzzHandshakeTimeout = 3000 * time.Millisecond ) -// 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) +// BzzSpec is the spec of the generic swarm handshake +var BzzSpec = &protocols.Spec{ + Name: "bzz", + Version: 4, + MaxMsgSize: 10 * 1024 * 1024, + Messages: []interface{}{ + HandshakeMsg{}, + }, } -/* -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) { +// DiscoverySpec is the spec for the bzz discovery subprotocols +var DiscoverySpec = &protocols.Spec{ + Name: "hive", + Version: 4, + MaxMsgSize: 10 * 1024 * 1024, + Messages: []interface{}{ + peersMsg{}, + subPeersMsg{}, + }, +} - // 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 +// Addr interface that peerPool needs +type Addr interface { + OverlayPeer + Over() []byte + Under() []byte + String() string + Update(OverlayAddr) OverlayAddr } -/* -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, - } +// Peer interface represents an live peer connection +type Peer interface { + Addr // the address of a peer + Conn // the live connection (protocols.Peer) + LastActive() time.Time // last time active +} - // 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 - } - }() +// Conn interface represents an live peer connection +type Conn interface { + ID() discover.NodeID // the key that uniquely identifies the Node for the peerPool + Handshake(context.Context, interface{}, func(interface{}) error) (interface{}, error) // can send messages + Send(interface{}) error // can send messages + Drop(error) // disconnect this peer + Run(func(interface{}) error) error // the run function to run a protocol + Off() OverlayAddr +} - // 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 - } +// BzzConfig captures the config params used by the hive +type BzzConfig struct { + OverlayAddr []byte // base address of the overlay network + UnderlayAddr []byte // node's underlay address + HiveParams *HiveParams + NetworkID uint64 +} + +// Bzz is the swarm protocol bundle +type Bzz struct { + *Hive + NetworkID uint64 + localAddr *BzzAddr + mtx sync.Mutex + handshakes map[discover.NodeID]*HandshakeMsg + streamerSpec *protocols.Spec + streamerRun func(*BzzPeer) error +} + +// NewBzz is the swarm protocol constructor +// arguments +// * bzz config +// * overlay driver +// * peer store +func NewBzz(config *BzzConfig, kad Overlay, store state.Store, streamerSpec *protocols.Spec, streamerRun func(*BzzPeer) error) *Bzz { + return &Bzz{ + Hive: NewHive(config.HiveParams, kad, store), + NetworkID: config.NetworkID, + localAddr: &BzzAddr{config.OverlayAddr, config.UnderlayAddr}, + handshakes: make(map[discover.NodeID]*HandshakeMsg), + streamerRun: streamerRun, + streamerSpec: streamerSpec, } } -// 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) +// UpdateLocalAddr updates underlayaddress of the running node +func (b *Bzz) UpdateLocalAddr(byteaddr []byte) *BzzAddr { + b.localAddr = b.localAddr.Update(&BzzAddr{ + UAddr: byteaddr, + OAddr: b.localAddr.OAddr, + }).(*BzzAddr) + return b.localAddr } -// 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 +// NodeInfo returns the node's overlay address +func (b *Bzz) NodeInfo() interface{} { + return b.localAddr.Address() +} + +// Protocols return the protocols swarm offers +// Bzz implements the node.Service interface +// * handshake/hive +// * discovery +func (b *Bzz) Protocols() []p2p.Protocol { + protocol := []p2p.Protocol{ + { + Name: BzzSpec.Name, + Version: BzzSpec.Version, + Length: BzzSpec.Length(), + Run: b.runBzz, + NodeInfo: b.NodeInfo, + }, + { + Name: DiscoverySpec.Name, + Version: DiscoverySpec.Version, + Length: DiscoverySpec.Length(), + Run: b.RunProtocol(DiscoverySpec, b.Hive.Run), + NodeInfo: b.Hive.NodeInfo, + PeerInfo: b.Hive.PeerInfo, + }, } - if msg.Size > ProtocolMaxMsgSize { - return fmt.Errorf("message too long: %v > %v", msg.Size, ProtocolMaxMsgSize) + if b.streamerSpec != nil && b.streamerRun != nil { + protocol = append(protocol, p2p.Protocol{ + Name: b.streamerSpec.Name, + Version: b.streamerSpec.Version, + Length: b.streamerSpec.Length(), + Run: b.RunProtocol(b.streamerSpec, b.streamerRun), + }) } - // 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) - } + return protocol +} + +// APIs returns the APIs offered by bzz +// * hive +// Bzz implements the node.Service interface +func (b *Bzz) APIs() []rpc.API { + return []rpc.API{{ + Namespace: "hive", + Version: "3.0", + Service: b.Hive, + }} +} - 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) +// RunProtocol is a wrapper for swarm subprotocols +// returns a p2p protocol run function that can be assigned to p2p.Protocol#Run field +// arguments: +// * p2p protocol spec +// * run function taking BzzPeer as argument +// this run function is meant to block for the duration of the protocol session +// on return the session is terminated and the peer is disconnected +// the protocol waits for the bzz handshake is negotiated +// the overlay address on the BzzPeer is set from the remote handshake +func (b *Bzz) RunProtocol(spec *protocols.Spec, run func(*BzzPeer) error) func(*p2p.Peer, p2p.MsgReadWriter) error { + return func(p *p2p.Peer, rw p2p.MsgReadWriter) error { + // wait for the bzz protocol to perform the handshake + handshake, _ := b.GetHandshake(p.ID()) + defer b.removeHandshake(p.ID()) + select { + case <-handshake.done: + case <-time.After(bzzHandshakeTimeout): + return fmt.Errorf("%08x: %s protocol timeout waiting for handshake on %08x", b.BaseAddr()[:4], spec.Name, p.ID().Bytes()[:4]) } - 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) + if handshake.err != nil { + return fmt.Errorf("%08x: %s protocol closed: %v", b.BaseAddr()[:4], spec.Name, handshake.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) + // the handshake has succeeded so construct the BzzPeer and run the protocol + peer := &BzzPeer{ + Peer: protocols.NewPeer(p, rw, spec), + localAddr: b.localAddr, + BzzAddr: handshake.peerAddr, + lastActive: time.Now(), } + return run(peer) + } +} - default: - // no other message is allowed - invalidMsgCounter.Inc(1) - return fmt.Errorf("invalid message code: %v", msg.Code) +// performHandshake implements the negotiation of the bzz handshake +// shared among swarm subprotocols +func (b *Bzz) performHandshake(p *protocols.Peer, handshake *HandshakeMsg) error { + ctx, cancel := context.WithTimeout(context.Background(), bzzHandshakeTimeout) + defer func() { + close(handshake.done) + cancel() + }() + rsh, err := p.Handshake(ctx, handshake, b.checkHandshake) + if err != nil { + handshake.err = err + return err } + handshake.peerAddr = rsh.(*HandshakeMsg).Addr 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, - }, +// runBzz is the p2p protocol run function for the bzz base protocol +// that negotiates the bzz handshake +func (b *Bzz) runBzz(p *p2p.Peer, rw p2p.MsgReadWriter) error { + handshake, _ := b.GetHandshake(p.ID()) + if !<-handshake.init { + return fmt.Errorf("%08x: bzz already started on peer %08x", b.localAddr.Over()[:4], ToOverlayAddr(p.ID().Bytes())[:4]) } - - err = p2p.Send(self.rw, statusMsg, handshake) + close(handshake.init) + defer b.removeHandshake(p.ID()) + peer := protocols.NewPeer(p, rw, BzzSpec) + err := b.performHandshake(peer, handshake) if err != nil { + log.Warn(fmt.Sprintf("%08x: handshake failed with remote peer %08x: %v", b.localAddr.Over()[:4], ToOverlayAddr(p.ID().Bytes())[:4], err)) + return err } - - // read and handle remote status - var msg p2p.Msg - msg, err = self.rw.ReadMsg() + // fail if we get another handshake + msg, err := rw.ReadMsg() if err != nil { return err } + msg.Discard() + return errors.New("received multiple handshakes") +} - 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) - } +// BzzPeer is the bzz protocol view of a protocols.Peer (itself an extension of p2p.Peer) +// implements the Peer interface and all interfaces Peer implements: Addr, OverlayPeer +type BzzPeer struct { + *protocols.Peer // represents the connection for online peers + localAddr *BzzAddr // local Peers address + *BzzAddr // remote address -> implements Addr interface = protocols.Peer + lastActive time.Time // time is updated whenever mutexes are releasing +} - if status.NetworkId != self.NetworkId { - return fmt.Errorf("network id mismatch: %d (!= %d)", status.NetworkId, self.NetworkId) +func NewBzzTestPeer(p *protocols.Peer, addr *BzzAddr) *BzzPeer { + return &BzzPeer{ + Peer: p, + localAddr: addr, + BzzAddr: NewAddrFromNodeID(p.ID()), } +} - if Version != status.Version { - return fmt.Errorf("protocol version mismatch: %d (!= %d)", status.Version, Version) - } +// Off returns the overlay peer record for offline persistence +func (p *BzzPeer) Off() OverlayAddr { + return p.BzzAddr +} - 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())) +// LastActive returns the time the peer was last active +func (p *BzzPeer) LastActive() time.Time { + return p.lastActive +} - 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 - } - } +/* + Handshake - 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 - } +* Version: 8 byte integer version of the protocol +* NetworkID: 8 byte integer network identifier +* Addr: the address advertised by the node including underlay and overlay connecctions +*/ +type HandshakeMsg struct { + Version uint64 + NetworkID uint64 + Addr *BzzAddr - // hive sets syncstate so sync should start after node added - log.Info(fmt.Sprintf("syncronisation request sent with %v", self.syncState)) - self.syncRequest() + // peerAddr is the address received in the peer handshake + peerAddr *BzzAddr - return nil + init chan bool + done chan struct{} + err error } -func (self *bzz) sync(state *syncState) error { - // syncer setup - if self.syncer != nil { - return errors.New("sync request can only be sent once") - } +// String pretty prints the handshake +func (bh *HandshakeMsg) String() string { + return fmt.Sprintf("Handshake: Version: %v, NetworkID: %v, Addr: %v", bh.Version, bh.NetworkID, bh.Addr) +} - 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)) +// Perform initiates the handshake and validates the remote handshake message +func (b *Bzz) checkHandshake(hs interface{}) error { + rhs := hs.(*HandshakeMsg) + if rhs.NetworkID != b.NetworkID { + return fmt.Errorf("network id mismatch %d (!= %d)", rhs.NetworkID, b.NetworkID) } - 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 + if rhs.Version != uint64(BzzSpec.Version) { + return fmt.Errorf("version mismatch %d (!= %d)", rhs.Version, BzzSpec.Version) } - log.Trace(fmt.Sprintf("syncer set for peer %v", self)) return nil } -func (self *bzz) String() string { - return self.remoteAddr.String() +// removeHandshake removes handshake for peer with peerID +// from the bzz handshake store +func (b *Bzz) removeHandshake(peerID discover.NodeID) { + b.mtx.Lock() + defer b.mtx.Unlock() + delete(b.handshakes, peerID) } -// 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), +// GetHandshake returns the bzz handhake that the remote peer with peerID sent +func (b *Bzz) GetHandshake(peerID discover.NodeID) (*HandshakeMsg, bool) { + b.mtx.Lock() + defer b.mtx.Unlock() + handshake, found := b.handshakes[peerID] + if !found { + handshake = &HandshakeMsg{ + Version: uint64(BzzSpec.Version), + NetworkID: b.NetworkID, + Addr: b.localAddr, + init: make(chan bool, 1), + done: make(chan struct{}), + } + // when handhsake is first created for a remote peer + // it is initialised with the init + handshake.init <- true + b.handshakes[peerID] = handshake } - return addr + + return handshake, found } -// outgoing messages -// send retrieveRequestMsg -func (self *bzz) retrieve(req *retrieveRequestMsgData) error { - return self.send(retrieveRequestMsg, req) +// BzzAddr implements the PeerAddr interface +type BzzAddr struct { + OAddr []byte + UAddr []byte } -// send storeRequestMsg -func (self *bzz) store(req *storeRequestMsgData) error { - return self.send(storeRequestMsg, req) +// Address implements OverlayPeer interface to be used in Overlay +func (a *BzzAddr) Address() []byte { + return a.OAddr } -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) +// Over returns the overlay address +func (a *BzzAddr) Over() []byte { + return a.OAddr } -// queue storeRequestMsg in request db -func (self *bzz) deliveryRequest(reqs []*syncRequest) error { - req := &deliveryRequestMsgData{ - Deliver: reqs, - } - return self.send(deliveryRequestMsg, req) +// Under returns the underlay address +func (a *BzzAddr) Under() []byte { + return a.UAddr } -// 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) +// ID returns the nodeID from the underlay enode address +func (a *BzzAddr) ID() discover.NodeID { + return discover.MustParseNode(string(a.UAddr)).ID } -// send paymentMsg -func (self *bzz) Pay(units int, promise swap.Promise) { - req := &paymentMsgData{uint(units), promise.(*chequebook.Cheque)} - self.payment(req) +// Update updates the underlay address of a peer record +func (a *BzzAddr) Update(na OverlayAddr) OverlayAddr { + return &BzzAddr{a.OAddr, na.(Addr).Under()} } -// send paymentMsg -func (self *bzz) payment(req *paymentMsgData) error { - return self.send(paymentMsg, req) +// String pretty prints the address +func (a *BzzAddr) String() string { + return fmt.Sprintf("%x <%s>", a.OAddr, a.UAddr) +} + +// RandomAddr is a utility method generating an address from a public key +func RandomAddr() *BzzAddr { + key, err := crypto.GenerateKey() + if err != nil { + panic("unable to generate key") + } + pubkey := crypto.FromECDSAPub(&key.PublicKey) + var id discover.NodeID + copy(id[:], pubkey[1:]) + return NewAddrFromNodeID(id) } -// sends peersMsg -func (self *bzz) peers(req *peersMsgData) error { - return self.send(peersMsg, req) +// NewNodeIDFromAddr transforms the underlay address to an adapters.NodeID +func NewNodeIDFromAddr(addr Addr) discover.NodeID { + log.Info(fmt.Sprintf("uaddr=%s", string(addr.Under()))) + node := discover.MustParseNode(string(addr.Under())) + return node.ID } -func (self *bzz) send(msg uint64, data interface{}) error { - if self.hive.blockWrite { - return fmt.Errorf("network write blocked") +// NewAddrFromNodeID constucts a BzzAddr from a discover.NodeID +// the overlay address is derived as the hash of the nodeID +func NewAddrFromNodeID(id discover.NodeID) *BzzAddr { + return &BzzAddr{ + OAddr: ToOverlayAddr(id.Bytes()), + UAddr: []byte(discover.NewNode(id, net.IP{127, 0, 0, 1}, 30303, 30303).String()), } - 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() +} + +// NewAddrFromNodeIDAndPort constucts a BzzAddr from a discover.NodeID and port uint16 +// the overlay address is derived as the hash of the nodeID +func NewAddrFromNodeIDAndPort(id discover.NodeID, host net.IP, port uint16) *BzzAddr { + return &BzzAddr{ + OAddr: ToOverlayAddr(id.Bytes()), + UAddr: []byte(discover.NewNode(id, host, port, port).String()), } - return err +} + +// ToOverlayAddr creates an overlayaddress from a byte slice +func ToOverlayAddr(id []byte) []byte { + return crypto.Keccak256(id) } |