move router code

router/connection.go

@@ -0,0 +1,555 @@
+package weave
+
+import (
+	"encoding/binary"
+	"encoding/gob"
+	"encoding/hex"
+	"fmt"
+	"log"
+	"net"
+	"strconv"
+	"time"
+)
+
+func NewRemoteConnection(from, to *Peer, tcpAddr string) *RemoteConnection {
+	return &RemoteConnection{
+		local:         from,
+		remote:        to,
+		remoteTCPAddr: tcpAddr}
+}
+
+func (conn *RemoteConnection) Local() *Peer {
+	return conn.local
+}
+
+func (conn *RemoteConnection) Remote() *Peer {
+	return conn.remote
+}
+
+func (conn *RemoteConnection) RemoteTCPAddr() string {
+	return conn.remoteTCPAddr
+}
+
+func (conn *RemoteConnection) Shutdown() {
+}
+
+func (conn *RemoteConnection) Established() bool {
+	return true
+}
+
+func (conn *RemoteConnection) String() string {
+	from := "<nil>"
+	if conn.local != nil {
+		from = conn.local.Name.String()
+	}
+	to := "<nil>"
+	if conn.remote != nil {
+		to = conn.remote.Name.String()
+	}
+	return fmt.Sprint("Connection ", from, "->", to)
+}
+
+// Async. Does not return anything. If the connection is successful,
+// it will end up in the local peer's connections map.
+func NewLocalConnection(connRemote *RemoteConnection, acceptNewPeer bool, tcpConn *net.TCPConn, udpAddr *net.UDPAddr, router *Router) {
+	if connRemote.local != router.Ourself {
+		log.Fatal("Attempt to create local connection from a peer which is not ourself")
+	}
+
+	queryChan := make(chan *ConnectionInteraction, ChannelSize)
+	// NB, we're taking a copy of connRemote here.
+	connLocal := &LocalConnection{
+		RemoteConnection: *connRemote,
+		Router:           router,
+		TCPConn:          tcpConn,
+		remoteUDPAddr:    udpAddr,
+		effectivePMTU:    DefaultPMTU,
+		queryChan:        queryChan}
+	go connLocal.queryLoop(queryChan, acceptNewPeer)
+}
+
+func (conn *LocalConnection) Established() bool {
+	conn.RLock()
+	defer conn.RUnlock()
+	return conn.established
+}
+
+// Read by the forwarder processes when in the UDP senders
+func (conn *LocalConnection) RemoteUDPAddr() *net.UDPAddr {
+	conn.RLock()
+	defer conn.RUnlock()
+	return conn.remoteUDPAddr
+}
+
+// Called by the forwarder processes in a few places (including
+// crypto), but the connection TCP receiver process, and by the local
+// peer actor process. Do not call this from the connection's actor
+// process itself.
+func (conn *LocalConnection) CheckFatal(err error) error {
+	if err == nil {
+		return nil
+	}
+	conn.log("encountered fatal error", err)
+	conn.Shutdown()
+	return err
+}
+
+// Called by forwarder processes, read in Forward (by sniffer and udp
+// listener process in router).
+func (conn *LocalConnection) setEffectivePMTU(pmtu int) {
+	conn.Lock()
+	defer conn.Unlock()
+	if conn.effectivePMTU != pmtu {
+		conn.effectivePMTU = pmtu
+		conn.log("Effective PMTU set to", pmtu)
+	}
+}
+
+// Called by the connection's actor process, by the connection's
+// heartbeat process and by the connection's TCP received
+// process. StackFrag is read in conn.Forward (called by router udp
+// listener and sniffer processes)
+func (conn *LocalConnection) setStackFrag(frag bool) {
+	conn.Lock()
+	defer conn.Unlock()
+	conn.stackFrag = frag
+}
+
+func (conn *LocalConnection) log(args ...interface{}) {
+	// why oh why can't I just write
+	// log.Println("Connection to peer", conn.remote.Name, args...)?
+	v := append([]interface{}{}, fmt.Sprintf("->[%s]:", conn.remote.Name))
+	v = append(v, args...)
+	log.Println(v...)
+}
+
+// ACTOR client API
+
+const (
+	CSendTCP          = iota
+	CSetEstablished   = iota
+	CSetRemoteUDPAddr = iota
+	CShutdown         = iota
+)
+
+// Async
+func (conn *LocalConnection) Shutdown() {
+	conn.queryChan <- &ConnectionInteraction{Interaction: Interaction{code: CShutdown}}
+}
+
+// Async
+func (conn *LocalConnection) SetRemoteUDPAddr(remoteUDPAddr *net.UDPAddr) {
+	if remoteUDPAddr == nil {
+		return
+	}
+	conn.queryChan <- &ConnectionInteraction{
+		Interaction: Interaction{code: CSetRemoteUDPAddr},
+		payload:     remoteUDPAddr}
+}
+
+// Async
+func (conn *LocalConnection) SetEstablished() {
+	conn.queryChan <- &ConnectionInteraction{Interaction: Interaction{code: CSetEstablished}}
+}
+
+// Async
+func (conn *LocalConnection) SendTCP(msg []byte) {
+	conn.queryChan <- &ConnectionInteraction{
+		Interaction: Interaction{code: CSendTCP},
+		payload:     msg}
+}
+
+// ACTOR server
+
+func (conn *LocalConnection) queryLoop(queryChan <-chan *ConnectionInteraction, acceptNewPeer bool) {
+	err := conn.handshake(acceptNewPeer)
+	if err != nil {
+		log.Printf("->[%s] encountered error during handshake: %v\n", conn.remoteTCPAddr, err)
+		conn.handleShutdown()
+		return
+	}
+	conn.local.AddConnection(conn)
+	if conn.remoteUDPAddr != nil {
+		err = conn.ensureHeartbeat(true)
+	}
+	terminate := false
+	for !terminate {
+		if err != nil {
+			conn.log("encountered fatal error", err)
+			break
+		}
+		query, ok := <-queryChan
+		if !ok {
+			break
+		}
+		switch query.code {
+		case CShutdown:
+			terminate = true
+		case CSetEstablished:
+			err = conn.handleSetEstablished()
+		case CSetRemoteUDPAddr:
+			err = conn.handleSetRemoteUDPAddr(query.payload.(*net.UDPAddr))
+		case CSendTCP:
+			err = conn.handleSendTCP(query.payload.([]byte))
+		}
+	}
+	conn.handleShutdown()
+}
+
+func (conn *LocalConnection) handleSetRemoteUDPAddr(remoteUDPAddr *net.UDPAddr) error {
+	if remoteUDPAddr == nil {
+		return nil
+	}
+	conn.Lock()
+	old := conn.remoteUDPAddr
+	conn.remoteUDPAddr = remoteUDPAddr
+	conn.Unlock()
+	if old == nil {
+		if err := conn.handleSendTCP(ProtocolConnectionEstablishedByte); err != nil {
+			return err
+		}
+		return conn.handleSetEstablished()
+	} else if old.String() != remoteUDPAddr.String() {
+		log.Println("Peer", conn.remote.Name, "moved from", old, "to", remoteUDPAddr)
+	}
+	return nil
+}
+
+func (conn *LocalConnection) handleSetEstablished() error {
+	conn.Lock()
+	old := conn.established
+	conn.established = true
+	conn.Unlock()
+	if !old {
+		conn.local.ConnectionEstablished(conn)
+		if err := conn.ensureHeartbeat(false); err != nil {
+			return err
+		}
+		// Send a large frame down the DF channel in order to prompt
+		// PMTU discovery to start.
+		conn.Forward(true, &ForwardedFrame{
+			srcPeer: conn.local,
+			dstPeer: conn.remote,
+			frame:   PMTUDiscovery},
+			nil)
+		conn.setStackFrag(false)
+		return conn.handleSendTCP(ProtocolStartFragmentationTestByte)
+	}
+	return nil
+}
+
+func (conn *LocalConnection) handleSendTCP(msg []byte) error {
+	return conn.tcpSender.Send(msg)
+}
+
+func (conn *LocalConnection) handleShutdown() {
+	if conn.remote != nil {
+		conn.log("shutting down")
+	}
+
+	// Whilst some of these elements may have been written to whilst
+	// holding locks, they were only written to by the connection
+	// actor process. handleShutdown is only called by the connection
+	// actor process. So there is no need to take locks to read these
+	// (or write elements which are only read by the same
+	// process). Taking locks is only done for elements which are read
+	// by other processes.
+	if conn.shutdown {
+		return
+	}
+	conn.shutdown = true
+
+	if conn.TCPConn != nil {
+		checkWarn(conn.TCPConn.Close())
+	}
+
+	if conn.remote != nil {
+		conn.remote.DecrementLocalRefCount()
+		conn.local.DeleteConnection(conn)
+	}
+
+	if conn.heartbeatStop != nil {
+		// heartbeatStop is 0 length, so this send will synchronise
+		// with the receive.
+		conn.heartbeatStop <- nil
+	}
+
+	// blank out the forwardChan so that the router processes don't
+	// try to send any more
+	conn.stopForwarders()
+
+	conn.Router.ConnectionMaker.ConnectionTerminated(conn.remoteTCPAddr)
+}
+
+func (conn *LocalConnection) handshake(acceptNewPeer bool) error {
+	// We do not need to worry about locking in here as at this point,
+	// the connection is not reachable by any go-routine other than
+	// ourself. Only when we add this connection to the conn.local
+	// peer will it be visible from multiple go-routines.
+	tcpConn := conn.TCPConn
+	tcpConn.SetKeepAlive(true)
+	tcpConn.SetLinger(0)
+	tcpConn.SetNoDelay(true)
+
+	enc := gob.NewEncoder(tcpConn)
+
+	localConnID := randUint64()
+
+	versionStr := fmt.Sprint(ProtocolVersion)
+	handshakeSend := map[string]string{
+		"Protocol":        Protocol,
+		"ProtocolVersion": versionStr,
+		"PeerNameFlavour": PeerNameFlavour,
+		"Name":            conn.local.Name.String(),
+		"UID":             fmt.Sprint(conn.local.UID),
+		"ConnID":          fmt.Sprint(localConnID)}
+	handshakeRecv := map[string]string{}
+
+	usingPassword := conn.Router.UsingPassword()
+	var public, private *[32]byte
+	var err error
+	if usingPassword {
+		public, private, err = GenerateKeyPair()
+		if err != nil {
+			return err
+		}
+		handshakeSend["PublicKey"] = hex.EncodeToString(public[:])
+	}
+	enc.Encode(handshakeSend)
+
+	dec := gob.NewDecoder(tcpConn)
+	err = dec.Decode(&handshakeRecv)
+	if err != nil {
+		return err
+	}
+	_, err = checkHandshakeStringField("Protocol", Protocol, handshakeRecv)
+	if err != nil {
+		return err
+	}
+	_, err = checkHandshakeStringField("ProtocolVersion", versionStr, handshakeRecv)
+	if err != nil {
+		return err
+	}
+	_, err = checkHandshakeStringField("PeerNameFlavour", PeerNameFlavour, handshakeRecv)
+	if err != nil {
+		return err
+	}
+	nameStr, err := checkHandshakeStringField("Name", "", handshakeRecv)
+	if err != nil {
+		return err
+	}
+	name, err := PeerNameFromString(nameStr)
+	if err != nil {
+		return err
+	}
+	if !acceptNewPeer {
+		if _, found := conn.Router.Peers.Fetch(name); !found {
+			return fmt.Errorf("Found unknown remote name: %s at %s", name, conn.remoteTCPAddr)
+		}
+	}
+
+	uidStr, err := checkHandshakeStringField("UID", "", handshakeRecv)
+	if err != nil {
+		return err
+	}
+	uid, err := strconv.ParseUint(uidStr, 10, 64)
+	if err != nil {
+		return err
+	}
+
+	remoteConnIdStr, err := checkHandshakeStringField("ConnID", "", handshakeRecv)
+	if err != nil {
+		return err
+	}
+	remoteConnID, err := strconv.ParseUint(remoteConnIdStr, 10, 64)
+	if err != nil {
+		return err
+	}
+	conn.UID = localConnID ^ remoteConnID
+
+	if usingPassword {
+		remotePublicStr, rpErr := checkHandshakeStringField("PublicKey", "", handshakeRecv)
+		if rpErr != nil {
+			return rpErr
+		}
+		remotePublicSlice, rpErr := hex.DecodeString(remotePublicStr)
+		if rpErr != nil {
+			return rpErr
+		}
+		remotePublic := [32]byte{}
+		for idx, elem := range remotePublicSlice {
+			remotePublic[idx] = elem
+		}
+		conn.SessionKey = FormSessionKey(&remotePublic, private, conn.Router.Password)
+		conn.tcpSender = NewEncryptedTCPSender(enc, conn)
+		conn.Decryptor = NewNaClDecryptor(conn)
+	} else {
+		if _, found := handshakeRecv["PublicKey"]; found {
+			return fmt.Errorf("Remote network is encrypted. Password required.")
+		}
+		conn.tcpSender = NewSimpleTCPSender(enc)
+		conn.Decryptor = NewNonDecryptor(conn)
+	}
+
+	toPeer := NewPeer(name, uid, 0, conn.Router)
+	toPeer = conn.Router.Peers.FetchWithDefault(toPeer)
+	if toPeer == nil {
+		return fmt.Errorf("Connection appears to be with different version of a peer we already know of")
+	} else if toPeer == conn.local {
+		// have to do assigment here to ensure Shutdown releases ref count
+		conn.remote = toPeer
+		return fmt.Errorf("Cannot connect to ourself")
+	}
+	conn.remote = toPeer
+
+	go conn.receiveTCP(dec, usingPassword)
+	return nil
+}
+
+func checkHandshakeStringField(fieldName string, expectedValue string, handshake map[string]string) (string, error) {
+	val, found := handshake[fieldName]
+	if !found {
+		return "", fmt.Errorf("Field % is missing", fieldName)
+	}
+	if expectedValue != "" && val != expectedValue {
+		return "", fmt.Errorf("Field %s has wrong value; expected '%s', received '%s'", fieldName, expectedValue, val)
+	}
+	return val, nil
+}
+
+func (conn *LocalConnection) receiveTCP(decoder *gob.Decoder, usingPassword bool) {
+	defer conn.Decryptor.Shutdown()
+	var receiver TCPReceiver
+	if usingPassword {
+		receiver = NewEncryptedTCPReceiver(conn)
+	} else {
+		receiver = NewSimpleTCPReceiver()
+	}
+	var err error
+	for {
+		var msg []byte
+		if conn.CheckFatal(decoder.Decode(&msg)) != nil {
+			return
+		}
+		msg, err = receiver.Decode(msg)
+		if err != nil {
+			checkWarn(err)
+			// remote peer may be using wrong password. Or there could
+			// be some sort of injection attack going on. Just ignore
+			// the traffic rather than shutting down.
+			continue
+		}
+		if msg[0] == ProtocolConnectionEstablished {
+			// We initiated the connection. We sent fast heartbeats to
+			// the remote side, which has now received at least one of
+			// them and thus has informed us via TCP that it considers
+			// the connection is now up. We now do a fetchAll on it.
+			conn.SetEstablished()
+			conn.SendTCP(ProtocolFetchAllByte)
+		} else if msg[0] == ProtocolStartFragmentationTest {
+			conn.Forward(false, &ForwardedFrame{
+				srcPeer: conn.local,
+				dstPeer: conn.remote,
+				frame:   FragTest},
+				nil)
+		} else if msg[0] == ProtocolFragmentationReceived {
+			conn.setStackFrag(true)
+		} else if usingPassword && msg[0] == ProtocolNonce {
+			conn.Decryptor.ReceiveNonce(msg[1:])
+		} else if msg[0] == ProtocolFetchAll {
+			// There are exactly two messages that relate to topology
+			// updates.
+			//
+			// 1. FetchAll. This carries no payload. The receiver
+			// responds with the entire topology model as the receiver
+			// has it.
+			//
+			// 2. Update. This carries a topology payload. The
+			// receiver merges it with its own topology model. If the
+			// payload is a subset of the receiver's topology, no
+			// further action is taken. Otherwise, the receiver sends
+			// out to all its connections an "improved" update:
+			//  - elements which the original payload added to the
+			//    receiver are included
+			//  - elements which the original payload updated in the
+			//    receiver are included
+			//  - elements which are equal between the receiver and
+			//    the payload are not included
+			//  - elements where the payload was older than the
+			//    receiver's version are updated
+			conn.SendTCP(Concat(ProtocolUpdateByte, conn.Router.Topology.FetchAll()))
+		} else if msg[0] == ProtocolUpdate {
+			newUpdate, err := conn.Router.Peers.ApplyUpdate(msg[1:], conn.Router)
+			if _, ok := err.(UnknownPeersError); err != nil && ok {
+				// That update contained a peer we didn't know about;
+				// request full update
+				conn.SendTCP(ProtocolFetchAllByte)
+				continue
+			} else if conn.CheckFatal(err) != nil {
+				return
+			}
+			if len(newUpdate) == 0 {
+				continue
+			}
+			conn.local.BroadcastTCP(Concat(ProtocolUpdateByte, newUpdate))
+		} else if msg[0] == ProtocolPMTUVerified {
+			conn.verifyPMTU <- int(binary.BigEndian.Uint16(msg[1:]))
+		} else {
+			conn.log("received unknown msg:\n", msg)
+		}
+	}
+}
+
+// Heartbeats
+
+// Heartbeating serves two purposes: a) keeping NAT paths alive, and
+// b) updating a remote peer's knowledge of our address, in the event
+// it changes (e.g. because NAT paths expired).
+// Called only by connection actor process.
+func (conn *LocalConnection) ensureHeartbeat(fast bool) error {
+	if err := conn.ensureForwarders(); err != nil {
+		return err
+	}
+	var heartbeat, fetchAll, fragTest <-chan time.Time
+	// explicitly 0 length chan - make send block until receive occurs
+	stop := make(chan interface{}, 0)
+	if fast {
+		// fast, nofetchall, no fragtest
+		// Lang Spec: "A nil channel is never ready for communication."
+		heartbeat = time.Tick(FastHeartbeat)
+	} else {
+		heartbeat = time.Tick(SlowHeartbeat)
+		fetchAll = time.Tick(FetchAllInterval)
+		fragTest = time.Tick(FragTestInterval)
+	}
+	// Don't need locks here as this is only read here and in
+	// handleShutdown, both of which are called by the connection
+	// actor process only.
+	if conn.heartbeatStop != nil {
+		conn.heartbeatStop <- nil
+	}
+	conn.heartbeatStop = stop
+	go conn.forwardHeartbeats(heartbeat, fetchAll, fragTest, stop)
+	return nil
+}
+
+func (conn *LocalConnection) forwardHeartbeats(heartbeat, fetchAll, fragTest <-chan time.Time, stop <-chan interface{}) {
+	heartbeatFrame := &ForwardedFrame{
+		srcPeer: conn.local,
+		dstPeer: conn.remote,
+		frame:   []byte{}}
+	conn.Forward(true, heartbeatFrame, nil) // avoid initial wait
+	for {
+		select {
+		case <-stop:
+			return
+		case <-heartbeat:
+			conn.Forward(true, heartbeatFrame, nil)
+		case <-fetchAll:
+			conn.SendTCP(ProtocolFetchAllByte)
+		case <-fragTest:
+			conn.setStackFrag(false)
+			conn.SendTCP(ProtocolStartFragmentationTestByte)
+		}
+	}
+}

router/connection_maker.go

@@ -0,0 +1,199 @@
+package weave
+
+import (
+	"bytes"
+	"fmt"
+	"log"
+	"math/rand"
+	"net"
+	"time"
+)
+
+const (
+	InitialInterval = 5 * time.Second
+	MaxInterval     = 10 * time.Minute
+)
+
+const (
+	CMInitiate   = iota
+	CMTerminated = iota
+	CMRefresh    = iota
+	CMStatus     = iota
+)
+
+func StartConnectionMaker(router *Router) *ConnectionMaker {
+	queryChan := make(chan *ConnectionMakerInteraction, ChannelSize)
+	state := &ConnectionMaker{
+		router:         router,
+		queryChan:      queryChan,
+		cmdLineAddress: make(map[string]bool),
+		targets:        make(map[string]*Target)}
+	go state.queryLoop(queryChan)
+	return state
+}
+
+func (cm *ConnectionMaker) InitiateConnection(address string) {
+	cm.queryChan <- &ConnectionMakerInteraction{
+		Interaction: Interaction{code: CMInitiate},
+		address:     address}
+}
+
+func (cm *ConnectionMaker) ConnectionTerminated(address string) {
+	cm.queryChan <- &ConnectionMakerInteraction{
+		Interaction: Interaction{code: CMTerminated},
+		address:     address}
+}
+
+func (cm *ConnectionMaker) Refresh() {
+	cm.queryChan <- &ConnectionMakerInteraction{
+		Interaction: Interaction{code: CMRefresh}}
+}
+
+func (cm *ConnectionMaker) String() string {
+	resultChan := make(chan interface{}, 0)
+	cm.queryChan <- &ConnectionMakerInteraction{
+		Interaction: Interaction{code: CMStatus, resultChan: resultChan}}
+	result := <-resultChan
+	return result.(string)
+}
+
+func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteraction) {
+	timer := time.NewTimer(MaxDuration)
+	run := func() { timer.Reset(cm.checkStateAndAttemptConnections()) }
+	for {
+		select {
+		case query, ok := <-queryChan:
+			if !ok {
+				return
+			}
+			switch query.code {
+			case CMInitiate:
+				cm.cmdLineAddress[NormalisePeerAddr(query.address)] = true
+				run()
+			case CMTerminated:
+				if target, found := cm.targets[query.address]; found {
+					target.attempting = false
+					target.tryAfter, target.tryInterval = tryAfter(target.tryInterval)
+				}
+				run()
+			case CMRefresh:
+				run()
+			case CMStatus:
+				run()
+				query.resultChan <- cm.status()
+			default:
+				log.Fatal("Unexpected connection maker query:", query)
+			}
+		case <-timer.C:
+			run()
+		}
+	}
+}
+
+func (cm *ConnectionMaker) checkStateAndAttemptConnections() time.Duration {
+	ourself := cm.router.Ourself
+	validTarget := make(map[string]bool)
+
+	// copy the set of things we are connected to, so we can access them without locking
+	our_connected_peers := make(map[PeerName]bool)
+	our_connected_targets := make(map[string]bool)
+	ourself.ForEachConnection(func(peer PeerName, conn Connection) {
+		our_connected_peers[peer] = true
+		our_connected_targets[conn.RemoteTCPAddr()] = true
+	})
+
+	addTarget := func(address string) {
+		if !our_connected_targets[address] {
+			validTarget[address] = true
+			cm.addTarget(address)
+		}
+	}
+
+	// Add command-line targets that are not connected
+	for address, _ := range cm.cmdLineAddress {
+		addTarget(address)
+	}
+
+	// Add targets for peers that someone else is connected to, but we
+	// aren't
+	cm.router.Peers.ForEach(func(name PeerName, peer *Peer) {
+		peer.ForEachConnection(func(otherPeer PeerName, conn Connection) {
+			if otherPeer == ourself.Name || our_connected_peers[otherPeer] {
+				return
+			}
+			address := conn.RemoteTCPAddr()
+			// try both portnumber of connection and standard port
+			addTarget(address)
+			if host, _, err := net.SplitHostPort(address); err == nil {
+				addTarget(NormalisePeerAddr(host))
+			}
+		})
+	})
+
+	now := time.Now() // make sure we catch items just added
+	after := MaxDuration
+	for address, target := range cm.targets {
+		if our_connected_targets[address] {
+			delete(cm.targets, address)
+			continue
+		}
+		if target.attempting {
+			continue
+		}
+		if !validTarget[address] {
+			delete(cm.targets, address)
+			continue
+		}
+		switch duration := target.tryAfter.Sub(now); {
+		case duration <= 0:
+			target.attempting = true
+			go cm.attemptConnection(address, cm.cmdLineAddress[address])
+		case duration < after:
+			after = duration
+		}
+	}
+	return after
+}
+
+func (cm *ConnectionMaker) addTarget(address string) {
+	if _, found := cm.targets[address]; !found {
+		target := &Target{}
+		target.tryAfter, target.tryInterval = tryImmediately()
+		cm.targets[address] = target
+	}
+}
+
+func (cm *ConnectionMaker) status() string {
+	var buf bytes.Buffer
+	for address, target := range cm.targets {
+		var fmtStr string
+		if target.attempting {
+			fmtStr = "%s (trying since %v)\n"
+		} else {
+			fmtStr = "%s (next try at %v)\n"
+		}
+		buf.WriteString(fmt.Sprintf(fmtStr, address, target.tryAfter))
+	}
+	return buf.String()
+}
+
+func (cm *ConnectionMaker) attemptConnection(address string, acceptNewPeer bool) {
+	log.Println("Attempting connection to", address)
+	if err := cm.router.Ourself.CreateConnection(address, acceptNewPeer); err != nil {
+		log.Println(err)
+		cm.ConnectionTerminated(address)
+	}
+}
+
+func tryImmediately() (time.Time, time.Duration) {
+	interval := time.Duration(rand.Int63n(int64(InitialInterval)))
+	return time.Now(), interval
+}
+
+func tryAfter(interval time.Duration) (time.Time, time.Duration) {
+	interval += time.Duration(rand.Int63n(int64(interval)))
+	if interval > MaxInterval {
+		interval = MaxInterval
+	}
+	return time.Now().Add(interval), interval
+}

router/consts.go

@@ -0,0 +1,50 @@
+package weave
+
+import (
+	"time"
+	"math"
+)
+
+const (
+	Protocol           = "weave"
+	ProtocolVersion    = 8
+	EthernetOverhead   = 14
+	UDPOverhead        = 28 // 20 bytes for IPv4, 8 bytes for UDP
+	Port               = 6783
+	StatusPort         = Port + 1
+	DefaultPMTU        = 65535
+	MaxUDPPacketSize   = 65536
+	ChannelSize        = 16
+	UDPNonceSendAt     = 8192
+	FragTestSize       = 60001
+	PMTUDiscoverySize  = 60000
+	FastHeartbeat      = 500 * time.Millisecond
+	SlowHeartbeat      = 10 * time.Second
+	FetchAllInterval   = 30 * time.Second
+	FragTestInterval   = 5 * time.Minute
+	PMTUVerifyAttempts = 8
+	PMTUVerifyTimeout  = 10 * time.Millisecond // gets doubled with every attempt
+	MaxDuration        = time.Duration(math.MaxInt64)
+)
+
+const (
+	ProtocolConnectionEstablished  = iota
+	ProtocolFragmentationReceived  = iota
+	ProtocolStartFragmentationTest = iota
+	ProtocolNonce                  = iota
+	ProtocolFetchAll               = iota
+	ProtocolUpdate                 = iota
+	ProtocolPMTUVerified           = iota
+)
+
+var (
+	FragTest                           = make([]byte, FragTestSize)
+	PMTUDiscovery                      = make([]byte, PMTUDiscoverySize)
+	ProtocolConnectionEstablishedByte  = []byte{ProtocolConnectionEstablished}
+	ProtocolFragmentationReceivedByte  = []byte{ProtocolFragmentationReceived}
+	ProtocolStartFragmentationTestByte = []byte{ProtocolStartFragmentationTest}
+	ProtocolNonceByte                  = []byte{ProtocolNonce}
+	ProtocolFetchAllByte               = []byte{ProtocolFetchAll}
+	ProtocolUpdateByte                 = []byte{ProtocolUpdate}
+	ProtocolPMTUVerifiedByte           = []byte{ProtocolPMTUVerified}
+)

router/crypto.go

@@ -0,0 +1,449 @@
+package weave
+
+import (
+	"bytes"
+	"code.google.com/p/go-bit/bit"
+	"code.google.com/p/go.crypto/nacl/box"
+	"code.google.com/p/go.crypto/nacl/secretbox"
+	"crypto/rand"
+	"crypto/sha256"
+	"encoding/binary"
+	"encoding/gob"
+	"fmt"
+	"log"
+)
+
+func GenerateKeyPair() (publicKey, privateKey *[32]byte, err error) {
+	return box.GenerateKey(rand.Reader)
+}
+
+func FormSessionKey(remotePublicKey, localPrivateKey *[32]byte, secretKey *[]byte) *[32]byte {
+	var sharedKey [32]byte
+	box.Precompute(&sharedKey, remotePublicKey, localPrivateKey)
+	sharedKeySlice := sharedKey[:]
+	sharedKeySlice = append(sharedKeySlice, *secretKey...)
+	sessionKey := sha256.Sum256(sharedKeySlice)
+	return &sessionKey
+}
+
+func GenerateRandomNonce() ([24]byte, error) {
+	var nonce [24]byte
+	n, err := rand.Read(nonce[:])
+	if err != nil {
+		return nonce, err
+	}
+	if n != 24 {
+		return nonce, fmt.Errorf("Did not read enough - wanted 24, got %v", n)
+	}
+	return nonce, nil
+}
+
+func EncryptPrefixNonce(plaintxt []byte, nonce *[24]byte, secret *[32]byte) []byte {
+	buf := make([]byte, 24, 24+len(plaintxt)+secretbox.Overhead)
+	copy(buf, nonce[:])
+	// Seal *appends* to buf
+	return secretbox.Seal(buf, plaintxt, nonce, secret)
+}
+
+func DecryptPrefixNonce(ciphertxt []byte, secret *[32]byte) ([]byte, bool) {
+	if len(ciphertxt) < secretbox.Overhead+24 {
+		return nil, false
+	}
+	// There is no way to nicely convert from a slice to an
+	// array. So have to used the following loop.
+	var nonce [24]byte
+	for idx, e := range ciphertxt[0:24] {
+		nonce[idx] = e
+	}
+	ciphertxt = ciphertxt[24:]
+	return secretbox.Open(nil, ciphertxt, &nonce, secret)
+}
+
+func SetNonceLow15Bits(nonce *[24]byte, offset uint16) {
+	// ensure top bit of offset is 0
+	offset = offset & ((1 << 15) - 1)
+	// grab top bit of nonce[22:24] (and clear out lower bits)
+	nonceBits := binary.BigEndian.Uint16(nonce[22:24]) & (1 << 15)
+	// Big endian => the MSB is stored at the *lowest* address. So
+	// that top bit in nonce[22] should stay as the top bit in
+	// nonce[22]
+	binary.BigEndian.PutUint16(nonce[22:24], nonceBits+offset)
+}
+
+// Nonce encoding/decoding
+
+func EncodeNonce(df bool) (*[24]byte, []byte, error) {
+	nonce, err := GenerateRandomNonce()
+	if err != nil {
+		return nil, []byte{}, err
+	}
+	// wipe out lowest 15 bits, but encode the df right at the bottom
+	flags := uint16(0)
+	if df {
+		flags = flags | 1
+	}
+	SetNonceLow15Bits(&nonce, flags)
+	return &nonce, Concat(ProtocolNonceByte, nonce[:]), nil
+}
+
+func DecodeNonce(msg []byte) (bool, *[24]byte) {
+	flags := uint16(msg[23])
+	df := 0 != (flags & 1)
+	nonce := [24]byte{}
+	// upper bound is exclusive so this avoids copying the flags
+	for idx, elem := range msg[:23] {
+		nonce[idx] = elem
+	}
+	return df, &nonce
+}
+
+// Frame Encryptors
+
+func NewNonEncryptor(prefix []byte) *NonEncryptor {
+	buf := make([]byte, MaxUDPPacketSize)
+	prefixLen := copy(buf, prefix)
+	return &NonEncryptor{
+		buf:       buf,
+		bufTail:   buf[prefixLen:],
+		buffered:  prefixLen,
+		prefixLen: prefixLen}
+}
+
+func (ne *NonEncryptor) PacketOverhead() int {
+	return ne.prefixLen
+}
+
+func (ne *NonEncryptor) FrameOverhead() int {
+	return NameSize + NameSize + 2
+}
+
+func (ne *NonEncryptor) IsEmpty() bool {
+	return ne.buffered == ne.prefixLen
+}
+
+func (ne *NonEncryptor) Bytes() []byte {
+	buf := ne.buf[:ne.buffered]
+	ne.buffered = ne.prefixLen
+	ne.bufTail = ne.buf[ne.prefixLen:]
+	return buf
+}
+
+func (ne *NonEncryptor) AppendFrame(frame *ForwardedFrame) {
+	bufTail := ne.bufTail
+	srcLen := copy(bufTail, frame.srcPeer.NameByte)
+	bufTail = bufTail[srcLen:]
+	dstLen := copy(bufTail, frame.dstPeer.NameByte)
+	bufTail = bufTail[dstLen:]
+	frameLen := len(frame.frame)
+	binary.BigEndian.PutUint16(bufTail, uint16(frameLen))
+	bufTail = bufTail[2:]
+	copy(bufTail, frame.frame)
+	ne.bufTail = bufTail[frameLen:]
+	ne.buffered += srcLen + dstLen + 2 + frameLen
+}
+
+func (ne *NonEncryptor) TotalLen() int {
+	return ne.buffered
+}
+
+func NewNaClEncryptor(prefix []byte, conn *LocalConnection, df bool) *NaClEncryptor {
+	buf := make([]byte, MaxUDPPacketSize)
+	prefixLen := copy(buf, prefix)
+	flags := uint16(0)
+	if df {
+		flags = flags | (1 << 15)
+	}
+	return &NaClEncryptor{
+		NonEncryptor: *NewNonEncryptor([]byte{}),
+		buf:          buf,
+		offset:       0,
+		nonce:        nil,
+		nonceChan:    make(chan *[24]byte, ChannelSize),
+		flags:        flags,
+		prefixLen:    prefixLen,
+		conn:         conn,
+		df:           df}
+}
+
+func (ne *NaClEncryptor) Bytes() []byte {
+	plaintext := ne.NonEncryptor.Bytes()
+	offsetFlags := ne.offset | ne.flags
+	ciphertext := ne.buf
+	binary.BigEndian.PutUint16(ciphertext[ne.prefixLen:], offsetFlags)
+	nonce := ne.nonce
+	if nonce == nil {
+		freshNonce, encodedNonce, err := EncodeNonce(ne.df)
+		if err = ne.conn.CheckFatal(err); err != nil {
+			return []byte{}
+		}
+		ne.conn.SendTCP(encodedNonce)
+		ne.nonce = freshNonce
+		nonce = freshNonce
+	}
+	offset := ne.offset
+	SetNonceLow15Bits(nonce, offset)
+	// Seal *appends* to ciphertext
+	ciphertext = secretbox.Seal(ciphertext[:ne.prefixLen+2], plaintext, nonce, ne.conn.SessionKey)
+
+	offset = (offset + 1) & ((1 << 15) - 1)
+	if offset == 0 {
+		// need a new nonce please
+		ne.nonce = <-ne.nonceChan
+	} else if offset == 1<<14 { // half way through range, send new nonce
+		nonce, encodedNonce, err := EncodeNonce(ne.df)
+		if err = ne.conn.CheckFatal(err); err != nil {
+			return []byte{}
+		}
+		ne.nonceChan <- nonce
+		ne.conn.SendTCP(encodedNonce)
+	}
+	ne.offset = offset
+
+	return ciphertext
+}
+
+func (ne *NaClEncryptor) PacketOverhead() int {
+	return ne.prefixLen + 2 + secretbox.Overhead + ne.NonEncryptor.PacketOverhead()
+}
+
+func (ne *NaClEncryptor) TotalLen() int {
+	return ne.PacketOverhead() + ne.NonEncryptor.TotalLen()
+}
+
+// Frame Decryptors
+
+func NewNonDecryptor(conn *LocalConnection) *NonDecryptor {
+	return &NonDecryptor{conn: conn}
+}
+
+func (nd *NonDecryptor) IterateFrames(fun FrameConsumer, packet *UDPPacket) error {
+	buf := packet.Packet
+	for len(buf) >= (2 + NameSize + NameSize) {
+		srcNameByte := buf[:NameSize]
+		buf = buf[NameSize:]
+		dstNameByte := buf[:NameSize]
+		buf = buf[NameSize:]
+		length := binary.BigEndian.Uint16(buf[:2])
+		frame := buf[2 : 2+length]
+		buf = buf[2+length:]
+		err := fun(nd.conn, packet.Sender, srcNameByte, dstNameByte, length, frame)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (nd *NonDecryptor) Shutdown() {
+}
+
+func (nd *NonDecryptor) ReceiveNonce(msg []byte) {
+	log.Println("Received Nonce on non-encrypted channel. Ignoring.")
+}
+
+func NewNaClDecryptor(conn *LocalConnection) *NaClDecryptor {
+	inst := NaClDecryptorInstance{
+		nonce:               nil,
+		previousNonce:       nil,
+		usedOffsets:         bit.New(),
+		previousUsedOffsets: nil,
+		highestOffsetSeen:   0,
+		nonceChan:           make(chan *[24]byte, ChannelSize)}
+	instDF := NaClDecryptorInstance{
+		nonce:               nil,
+		previousNonce:       nil,
+		usedOffsets:         bit.New(),
+		previousUsedOffsets: nil,
+		highestOffsetSeen:   0,
+		nonceChan:           make(chan *[24]byte, ChannelSize)}
+	return &NaClDecryptor{
+		NonDecryptor: *NewNonDecryptor(conn),
+		instance:     &inst,
+		instanceDF:   &instDF}
+}
+
+func (nd *NaClDecryptor) Shutdown() {
+	close(nd.instance.nonceChan)
+	close(nd.instanceDF.nonceChan)
+}
+
+func (nd *NaClDecryptor) ReceiveNonce(msg []byte) {
+	df, nonce := DecodeNonce(msg)
+	if df {
+		nd.instanceDF.nonceChan <- nonce
+	} else {
+		nd.instance.nonceChan <- nonce
+	}
+}
+
+func (nd *NaClDecryptor) IterateFrames(fun FrameConsumer, packet *UDPPacket) error {
+	buf, err := nd.decrypt(packet.Packet)
+	if err != nil {
+		return err
+	}
+	packet.Packet = buf
+	return nd.NonDecryptor.IterateFrames(fun, packet)
+}
+
+func (nd *NaClDecryptor) decrypt(buf []byte) ([]byte, error) {
+	offset := binary.BigEndian.Uint16(buf[:2])
+	df := (offset & (1 << 15)) != 0
+	offsetNoFlags := offset & ((1 << 15) - 1)
+	var decState *NaClDecryptorInstance
+	if df {
+		decState = nd.instanceDF
+	} else {
+		decState = nd.instance
+	}
+	var nonce *[24]byte
+	var usedOffsets *bit.Set
+	var ok bool
+	if decState.nonce == nil {
+		if offsetNoFlags > (1 << 13) {
+			// offset is already beyond the first quarter and it's the
+			// first thing we've seen?! I don't think so.
+			return nil, fmt.Errorf("Unexpected offset when decrypting UDP packet")
+		}
+		decState.nonce, ok = <-decState.nonceChan
+		if !ok {
+			return nil, fmt.Errorf("Nonce chan closed")
+		}
+		nonce = decState.nonce
+		usedOffsets = decState.usedOffsets
+		decState.highestOffsetSeen = offsetNoFlags
+	} else {
+		highestOffsetSeen := decState.highestOffsetSeen
+		if offsetNoFlags < (1<<13) && highestOffsetSeen > ((1<<14)+(1<<13)) &&
+			(highestOffsetSeen-offsetNoFlags) > ((1<<14)+(1<<13)) {
+			// offset is in the first quarter, highestOffsetSeen is in
+			// the top quarter and under a quarter behind us. We
+			// interpret this as we need to move to the next nonce
+			decState.previousUsedOffsets = decState.usedOffsets
+			decState.usedOffsets = bit.New()
+			decState.previousNonce = decState.nonce
+			decState.nonce, ok = <-decState.nonceChan
+			if !ok {
+				return nil, fmt.Errorf("Nonce chan closed")
+			}
+			decState.highestOffsetSeen = offsetNoFlags
+			nonce = decState.nonce
+			usedOffsets = decState.usedOffsets
+		} else if offsetNoFlags > highestOffsetSeen &&
+			(offsetNoFlags-highestOffsetSeen) < (1<<13) {
+			// offset is under a quarter above highestOffsetSeen. This
+			// is ok - maybe some packet loss
+			decState.highestOffsetSeen = offsetNoFlags
+			nonce = decState.nonce
+			usedOffsets = decState.usedOffsets
+		} else if offsetNoFlags <= highestOffsetSeen &&
+			(highestOffsetSeen-offsetNoFlags) < (1<<13) {
+			// offset is within a quarter of the highest we've
+			// seen. This is ok - just assuming some out-of-order
+			// delivery.
+			nonce = decState.nonce
+			usedOffsets = decState.usedOffsets
+		} else if highestOffsetSeen < (1<<13) && offsetNoFlags > ((1<<14)+(1<<13)) &&
+			(offsetNoFlags-highestOffsetSeen) > ((1<<14)+(1<<13)) {
+			// offset is in the last quarter, highestOffsetSeen is in
+			// the first quarter, and offset is under a quarter behind
+			// us. This is ok - as above, just some out of order. But
+			// here it means we're dealing with the previous nonce
+			nonce = decState.previousNonce
+			usedOffsets = decState.previousUsedOffsets
+		} else {
+			return nil, fmt.Errorf("Unexpected offset when decrypting UDP packet")
+		}
+	}
+	offsetNoFlagsInt := int(offsetNoFlags)
+	if usedOffsets.Contains(offsetNoFlagsInt) {
+		return nil, fmt.Errorf("Suspected replay attack detected when decrypting UDP packet")
+	}
+	SetNonceLow15Bits(nonce, offsetNoFlags)
+	result, success := secretbox.Open(nil, buf[2:], nonce, nd.conn.SessionKey)
+	if success {
+		usedOffsets.Add(offsetNoFlagsInt)
+		return result, nil
+	} else {
+		return nil, fmt.Errorf("Unable to decrypt msg via UDP: %v", buf)
+	}
+}
+
+// TCP Senders
+
+func NewSimpleTCPSender(encoder *gob.Encoder) *SimpleTCPSender {
+	return &SimpleTCPSender{encoder: encoder}
+}
+
+func (sender *SimpleTCPSender) Send(msg []byte) error {
+	return sender.encoder.Encode(msg)
+}
+
+func NewEncryptedTCPSender(encoder *gob.Encoder, conn *LocalConnection) *EncryptedTCPSender {
+	buffer := new(bytes.Buffer)
+	return &EncryptedTCPSender{
+		outerEncoder: encoder,
+		innerEncoder: gob.NewEncoder(buffer),
+		buffer:       buffer,
+		conn:         conn,
+		msgCount:     0}
+}
+
+func (sender *EncryptedTCPSender) Send(msg []byte) error {
+	nonce, err := GenerateRandomNonce()
+	if err != nil {
+		return err
+	}
+	sender.Lock()
+	defer sender.Unlock()
+	wrappedMsg := EncryptedTCPMessage{Number: sender.msgCount, Body: msg}
+	buffer := sender.buffer
+	buffer.Reset()
+	err = sender.innerEncoder.Encode(wrappedMsg)
+	if err != nil {
+		return err
+	}
+	sender.msgCount = sender.msgCount + 1
+	return sender.outerEncoder.Encode(
+		EncryptPrefixNonce(buffer.Bytes(), &nonce, sender.conn.SessionKey))
+}
+
+// TCP Receivers
+
+func NewSimpleTCPReceiver() *SimpleTCPReceiver {
+	return &SimpleTCPReceiver{}
+}
+
+func (receiver *SimpleTCPReceiver) Decode(msg []byte) ([]byte, error) {
+	return msg, nil
+}
+
+func NewEncryptedTCPReceiver(conn *LocalConnection) *EncryptedTCPReceiver {
+	buffer := new(bytes.Buffer)
+	return &EncryptedTCPReceiver{
+		conn:     conn,
+		decoder:  gob.NewDecoder(buffer),
+		buffer:   buffer,
+		msgCount: 0}
+}
+
+func (receiver *EncryptedTCPReceiver) Decode(msg []byte) ([]byte, error) {
+	plaintext, success := DecryptPrefixNonce(msg, receiver.conn.SessionKey)
+	if !success {
+		return msg, fmt.Errorf("Unable to decrypt msg via TCP:\n %X", msg)
+	}
+	receiver.buffer.Reset()
+	_, err := receiver.buffer.Write(plaintext)
+	if err != nil {
+		return msg, err
+	}
+	wrappedMsg := new(EncryptedTCPMessage)
+	err = receiver.decoder.Decode(wrappedMsg)
+	if err != nil {
+		return msg, err
+	}
+	if wrappedMsg.Number != receiver.msgCount {
+		return msg, fmt.Errorf("Received TCP message with wrong sequence number; possible replay attack")
+	}
+	receiver.msgCount = receiver.msgCount + 1
+	return wrappedMsg.Body, nil
+}

router/ethernet_decoder.go

@@ -0,0 +1,91 @@
+package weave
+
+import (
+	"bytes"
+	"code.google.com/p/gopacket"
+	"code.google.com/p/gopacket/layers"
+	"log"
+	"net"
+)
+
+func NewEthernetDecoder() *EthernetDecoder {
+	dec := &EthernetDecoder{}
+	dec.parser = gopacket.NewDecodingLayerParser(layers.LayerTypeEthernet, &dec.eth, &dec.ip)
+	return dec
+}
+
+func (dec *EthernetDecoder) DecodeLayers(data []byte) error {
+	return dec.parser.DecodeLayers(data, &dec.decoded)
+}
+
+func (dec *EthernetDecoder) CheckFrameTooBig(err error, sendFrame func([]byte) error) error {
+	if ftbe, ok := err.(FrameTooBigError); ok {
+		// we know: 1. ip is valid, 2. it was ip and DF was set
+		icmpFrame, err := dec.formICMPMTUPacket(ftbe.EPMTU)
+		if err != nil {
+			return err
+		}
+		log.Printf("Sending ICMP 3,4 (%v -> %v): PMTU= %v\n", dec.ip.DstIP, dec.ip.SrcIP, ftbe.EPMTU)
+		return sendFrame(icmpFrame)
+	} else {
+		return err
+	}
+}
+
+func (dec *EthernetDecoder) formICMPMTUPacket(mtu int) ([]byte, error) {
+	buf := gopacket.NewSerializeBuffer()
+	opts := gopacket.SerializeOptions{
+		FixLengths:       true,
+		ComputeChecksums: true}
+	ipHeaderSize := int(dec.ip.IHL) * 4 // IHL is the number of 32-byte words in the header
+	payload := gopacket.Payload(dec.ip.BaseLayer.Contents[:ipHeaderSize+8])
+	err := gopacket.SerializeLayers(buf, opts,
+		&layers.Ethernet{
+			SrcMAC:       dec.eth.DstMAC,
+			DstMAC:       dec.eth.SrcMAC,
+			EthernetType: dec.eth.EthernetType},
+		&layers.IPv4{
+			Version:    4,
+			TOS:        dec.ip.TOS,
+			Id:         0,
+			Flags:      0,
+			FragOffset: 0,
+			TTL:        64,
+			Protocol:   layers.IPProtocolICMPv4,
+			DstIP:      dec.ip.SrcIP,
+			SrcIP:      dec.ip.DstIP},
+		&layers.ICMPv4{
+			TypeCode: 0x304,
+			Id:       0,
+			Seq:      uint16(mtu)},
+		&payload)
+	if err != nil {
+		return []byte{}, err
+	}
+	return buf.Bytes(), nil
+}
+
+var (
+	// see http://en.wikipedia.org/wiki/Multicast_address#Ethernet
+	broadcastMAC, _        = net.ParseMAC("ff:ff:ff:ff:ff:ff")
+	zeroMAC, _             = net.ParseMAC("00:00:00:00:00:00")
+	stpMACPrefix           = []byte{0x01, 0x80, 0xC2, 0x00, 0x00}
+	ipv4MulticastMACPrefix = []byte{0x01, 0x00, 0x5E}
+	ipv6MulticastMACPrefix = []byte{0x33, 0x33}
+)
+
+func (dec *EthernetDecoder) DropFrame() bool {
+	return bytes.Equal(stpMACPrefix, dec.eth.DstMAC[:len(stpMACPrefix)])
+}
+
+func (dec *EthernetDecoder) BroadcastFrame() bool {
+	return bytes.Equal(broadcastMAC, dec.eth.DstMAC) ||
+		// treat multicast frames as broadcast
+		bytes.Equal(ipv4MulticastMACPrefix, dec.eth.DstMAC[:len(ipv4MulticastMACPrefix)]) ||
+		bytes.Equal(ipv6MulticastMACPrefix, dec.eth.DstMAC[:len(ipv6MulticastMACPrefix)])
+}
+
+func (dec *EthernetDecoder) IsPMTUVerify() bool {
+	return bytes.Equal(zeroMAC, dec.eth.SrcMAC) &&
+		bytes.Equal(zeroMAC, dec.eth.DstMAC)
+}

router/forwarder.go

@@ -0,0 +1,443 @@
+package weave
+
+import (
+	"code.google.com/p/gopacket"
+	"code.google.com/p/gopacket/layers"
+	"log"
+	"net"
+	"syscall"
+	"time"
+)
+
+func (conn *LocalConnection) ensureForwarders() error {
+	if conn.forwardChan != nil || conn.forwardChanDF != nil {
+		return nil
+	}
+	udpSender := NewSimpleUDPSender(conn)
+	udpSenderDF, err := NewRawUDPSender(conn) // only thing that can error, so do it early
+	if err != nil {
+		return err
+	}
+
+	usingPassword := conn.SessionKey != nil
+	var encryptor, encryptorDF Encryptor
+	if usingPassword {
+		encryptor = NewNaClEncryptor(conn.local.NameByte, conn, false)
+		encryptorDF = NewNaClEncryptor(conn.local.NameByte, conn, true)
+	} else {
+		encryptor = NewNonEncryptor(conn.local.NameByte)
+		encryptorDF = NewNonEncryptor(conn.local.NameByte)
+	}
+
+	var (
+		forwardChan   = make(chan *ForwardedFrame, ChannelSize)
+		forwardChanDF = make(chan *ForwardedFrame, ChannelSize)
+		stopForward   = make(chan interface{}, 0)
+		stopForwardDF = make(chan interface{}, 0)
+		verifyPMTU    = make(chan int, ChannelSize)
+	)
+	//NB: only forwarderDF can ever encounter EMSGSIZE errors, and
+	//thus perform PMTU verification
+	forwarder := NewForwarder(conn, forwardChan, stopForward, nil, encryptor, udpSender, DefaultPMTU)
+	forwarderDF := NewForwarder(conn, forwardChanDF, stopForwardDF, verifyPMTU, encryptorDF, udpSenderDF, DefaultPMTU)
+
+	// Various fields in the conn struct are read by other processes,
+	// so we have to use locks.
+	conn.Lock()
+	conn.forwardChan = forwardChan
+	conn.forwardChanDF = forwardChanDF
+	conn.stopForward = stopForward
+	conn.stopForwardDF = stopForwardDF
+	conn.verifyPMTU = verifyPMTU
+	conn.effectivePMTU = forwarder.unverifiedPMTU
+	conn.Unlock()
+
+	go forwarder.Run()
+	go forwarderDF.Run()
+
+	return nil
+}
+
+func (conn *LocalConnection) stopForwarders() {
+	conn.Lock()
+	conn.forwardChan = nil
+	conn.forwardChanDF = nil
+	conn.Unlock()
+	// Now signal the forwarder loops to exit. They will drain the
+	// forwarder chans in order to unblock any router processes
+	// blocked on sending.
+	if conn.stopForward != nil {
+		conn.stopForward <- nil
+		conn.stopForwardDF <- nil
+	}
+}
+
+// Called from peer.Relay[Broadcast] which is itself invoked from
+// router (both UDP listener process and sniffer process). Also called
+// from connection's heartbeat process, and from the connection's TCP
+// receiver process.
+func (conn *LocalConnection) Forward(df bool, frame *ForwardedFrame, dec *EthernetDecoder) error {
+	conn.RLock()
+	var (
+		forwardChan   = conn.forwardChan
+		forwardChanDF = conn.forwardChanDF
+		effectivePMTU = conn.effectivePMTU
+		stackFrag     = conn.stackFrag
+	)
+	conn.RUnlock()
+
+	if forwardChan == nil || forwardChanDF == nil {
+		conn.log("Cannot forward frame yet - awaiting contact")
+		return nil
+	}
+	if df {
+		if !frameTooBig(frame, effectivePMTU) {
+			forwardChanDF <- frame
+			return nil
+		}
+		return FrameTooBigError{EPMTU: effectivePMTU}
+	} else {
+		if stackFrag || dec == nil || len(dec.decoded) < 2 {
+			forwardChan <- frame
+			return nil
+		}
+		// Don't have trustworthy stack, so we're going to have to
+		// send it DF in any case.
+		if !frameTooBig(frame, effectivePMTU) {
+			forwardChanDF <- frame
+			return nil
+		}
+		conn.Router.LogFrame("Fragmenting", frame.frame, &dec.eth)
+		// We can't trust the stack to fragment, we have IP, and we
+		// have a frame that's too big for the MTU, so we have to
+		// fragment it ourself.
+		return fragment(dec.eth, dec.ip, effectivePMTU, frame, func(segFrame *ForwardedFrame) {
+			forwardChanDF <- segFrame
+		})
+	}
+}
+
+func frameTooBig(frame *ForwardedFrame, effectivePMTU int) bool {
+	// We capture/forward complete ethernet frames. Therefore the
+	// frame length includes the ethernet header. However, MTUs
+	// operate at the IP layer and thus do not include the ethernet
+	// header. To put it another way, when a sender that was told an
+	// MTU of M sends an IP packet of exactly that length, we will
+	// capture/forward M + EthernetOverhead bytes of data.
+	return len(frame.frame) > effectivePMTU+EthernetOverhead
+}
+
+func fragment(eth layers.Ethernet, ip layers.IPv4, pmtu int, frame *ForwardedFrame, forward func(*ForwardedFrame)) error {
+	// We are not doing any sort of NAT, so we don't need to worry
+	// about checksums of IP payload (eg UDP checksum).
+	headerSize := int(ip.IHL) * 4
+	// &^ is bit clear (AND NOT). So here we're clearing the lowest 3
+	// bits.
+	maxSegmentSize := (pmtu - headerSize) &^ 7
+	opts := gopacket.SerializeOptions{
+		FixLengths:       false,
+		ComputeChecksums: true}
+	payloadSize := int(ip.Length) - headerSize
+	payload := ip.BaseLayer.Payload[:payloadSize]
+	offsetBase := int(ip.FragOffset) << 3
+	origFlags := ip.Flags
+	ip.Flags = ip.Flags | layers.IPv4MoreFragments
+	ip.Length = uint16(headerSize + maxSegmentSize)
+	if eth.EthernetType == layers.EthernetTypeLLC {
+		// using LLC, so must set eth length correctly. eth length
+		// is just the length of the payload
+		eth.Length = ip.Length
+	} else {
+		eth.Length = 0
+	}
+	for offset := 0; offset < payloadSize; offset += maxSegmentSize {
+		var segmentPayload []byte
+		if len(payload) <= maxSegmentSize {
+			// last one
+			segmentPayload = payload
+			ip.Length = uint16(len(payload) + headerSize)
+			ip.Flags = origFlags
+			if eth.EthernetType == layers.EthernetTypeLLC {
+				eth.Length = ip.Length
+			} else {
+				eth.Length = 0
+			}
+		} else {
+			segmentPayload = payload[:maxSegmentSize]
+			payload = payload[maxSegmentSize:]
+		}
+		ip.FragOffset = uint16((offset + offsetBase) >> 3)
+		buf := gopacket.NewSerializeBuffer()
+		segPayload := gopacket.Payload(segmentPayload)
+		err := gopacket.SerializeLayers(buf, opts, &eth, &ip, &segPayload)
+		if err != nil {
+			return err
+		}
+		// make copies of the frame we received
+		var segFrame ForwardedFrame = *frame
+		segFrame.frame = buf.Bytes()
+		forward(&segFrame)
+	}
+	return nil
+}
+
+// Forwarder
+
+func NewForwarder(conn *LocalConnection, ch <-chan *ForwardedFrame, stop <-chan interface{}, verifyPMTU <-chan int, enc Encryptor, udpSender UDPSender, pmtu int) *Forwarder {
+	fwd := &Forwarder{
+		conn:       conn,
+		ch:         ch,
+		stop:       stop,
+		verifyPMTU: verifyPMTU,
+		enc:        enc,
+		udpSender:  udpSender}
+	fwd.unverifiedPMTU = pmtu - fwd.effectiveOverhead()
+	fwd.maxPayload = pmtu - UDPOverhead
+	return fwd
+}
+
+func (fwd *Forwarder) Run() {
+	defer fwd.udpSender.Shutdown()
+	var flushed, ok bool
+	var frame *ForwardedFrame
+	for {
+		flushed = false
+		select {
+		case <-fwd.stop:
+			fwd.drain()
+			return
+		case <-fwd.verifyPMTUTick:
+			// We only do this case here when we know the buffers are
+			// all empty so that we don't risk appending verify-frames
+			// to other data.
+			fwd.verifyPMTUTick = nil
+			if fwd.pmtuVerified {
+				continue
+			}
+			if fwd.pmtuVerifyCount > 0 {
+				fwd.pmtuVerifyCount--
+				fwd.attemptVerifyEffectivePMTU()
+			} else {
+				// we've exceeded the verification attempts of the
+				// unverifiedPMTU
+				fwd.lowestBadPMTU = fwd.unverifiedPMTU
+				fwd.verifyEffectivePMTU((fwd.highestGoodPMTU + fwd.lowestBadPMTU) / 2)
+			}
+		case epmtu := <-fwd.verifyPMTU:
+			if fwd.pmtuVerified || epmtu != fwd.unverifiedPMTU {
+				continue
+			}
+			if epmtu+1 < fwd.lowestBadPMTU {
+				fwd.highestGoodPMTU = fwd.unverifiedPMTU // = epmtu
+				fwd.verifyEffectivePMTU((fwd.highestGoodPMTU + fwd.lowestBadPMTU) / 2)
+			} else {
+				fwd.pmtuVerified = true
+				fwd.maxPayload = epmtu + fwd.effectiveOverhead() - UDPOverhead
+				fwd.conn.setEffectivePMTU(epmtu)
+				fwd.conn.log("Effective PMTU verified at", epmtu)
+			}
+		case frame = <-fwd.ch:
+			if !fwd.appendFrame(frame) {
+				fwd.logDrop(frame)
+				continue
+			}
+			for !flushed {
+				select {
+				case frame, ok = <-fwd.ch:
+					if !ok {
+						return
+					}
+					if !fwd.appendFrame(frame) {
+						fwd.flush()
+						if !fwd.appendFrame(frame) {
+							fwd.logDrop(frame)
+							flushed = true
+						}
+					}
+				default:
+					fwd.flush()
+					flushed = true
+				}
+			}
+		}
+	}
+}
+
+func (fwd *Forwarder) effectiveOverhead() int {
+	return UDPOverhead + fwd.enc.PacketOverhead() + fwd.enc.FrameOverhead() + EthernetOverhead
+}
+
+func (fwd *Forwarder) verifyEffectivePMTU(newUnverifiedPMTU int) {
+	fwd.unverifiedPMTU = newUnverifiedPMTU
+	fwd.pmtuVerifyCount = PMTUVerifyAttempts
+	fwd.attemptVerifyEffectivePMTU()
+}
+
+func (fwd *Forwarder) attemptVerifyEffectivePMTU() {
+	pmtuVerifyFrame := &ForwardedFrame{
+		srcPeer: fwd.conn.local,
+		dstPeer: fwd.conn.remote,
+		frame:   make([]byte, fwd.unverifiedPMTU+EthernetOverhead)}
+	fwd.enc.AppendFrame(pmtuVerifyFrame)
+	fwd.flush()
+	if fwd.verifyPMTUTick == nil {
+		fwd.verifyPMTUTick = time.After(PMTUVerifyTimeout << (PMTUVerifyAttempts - fwd.pmtuVerifyCount))
+	}
+}
+
+func (fwd *Forwarder) appendFrame(frame *ForwardedFrame) bool {
+	frameLen := len(frame.frame)
+	if fwd.enc.TotalLen()+fwd.enc.FrameOverhead()+frameLen > fwd.maxPayload {
+		return false
+	}
+	fwd.enc.AppendFrame(frame)
+	return true
+}
+
+func (fwd *Forwarder) flush() {
+	err := fwd.udpSender.Send(fwd.enc.Bytes())
+	if err != nil {
+		if mtbe, ok := err.(MsgTooBigError); ok {
+			newUnverifiedPMTU := mtbe.PMTU - fwd.effectiveOverhead()
+			if newUnverifiedPMTU >= fwd.unverifiedPMTU {
+				return
+			}
+			fwd.pmtuVerified = false
+			fwd.maxPayload = mtbe.PMTU - UDPOverhead
+			fwd.highestGoodPMTU = 8
+			fwd.lowestBadPMTU = newUnverifiedPMTU + 1
+			fwd.conn.setEffectivePMTU(newUnverifiedPMTU)
+			fwd.verifyEffectivePMTU(newUnverifiedPMTU)
+		} else if PosixError(err) == syscall.ENOBUFS {
+			// TODO handle this better
+		} else {
+			fwd.conn.CheckFatal(err)
+		}
+	}
+}
+
+func (fwd *Forwarder) drain() {
+	// We want to drain before exiting otherwise we could get the
+	// packet sniffer or udp listener blocked on sending to a full
+	// chan
+	for {
+		select {
+		case <-fwd.ch:
+		default:
+			return
+		}
+	}
+}
+
+func (fwd *Forwarder) logDrop(frame *ForwardedFrame) {
+	fwd.conn.log("Dropping too big frame during forwarding: frame len:", len(frame.frame), "; effective PMTU:", fwd.maxPayload+UDPOverhead-fwd.effectiveOverhead())
+}
+
+// UDP Senders
+
+func NewSimpleUDPSender(conn *LocalConnection) *SimpleUDPSender {
+	return &SimpleUDPSender{udpConn: conn.Router.UDPListener, conn: conn}
+}
+
+func (sender *SimpleUDPSender) Send(msg []byte) error {
+	_, err := sender.udpConn.WriteToUDP(msg, sender.conn.RemoteUDPAddr())
+	return err
+}
+
+func (sender *SimpleUDPSender) Shutdown() error {
+	return nil
+}
+
+func NewRawUDPSender(conn *LocalConnection) (*RawUDPSender, error) {
+	ipSocket, err := dialIP(conn)
+	if err != nil {
+		return nil, err
+	}
+	udpHeader := &layers.UDP{SrcPort: layers.UDPPort(Port)}
+	ipBuf := gopacket.NewSerializeBuffer()
+	opts := gopacket.SerializeOptions{
+		FixLengths: true,
+		// UDP header is calculated with a phantom IP
+		// header. Yes, it's totally nuts. Thankfully, for UDP
+		// over IPv4, the checksum is optional. It's not
+		// optional for IPv6, but we'll ignore that for
+		// now. TODO
+		ComputeChecksums: false}
+
+	return &RawUDPSender{
+		ipBuf:     ipBuf,
+		opts:      opts,
+		udpHeader: udpHeader,
+		socket:    ipSocket,
+		conn:      conn}, nil
+}
+
+func (sender *RawUDPSender) Send(msg []byte) error {
+	payload := gopacket.Payload(msg)
+	sender.udpHeader.DstPort = layers.UDPPort(sender.conn.RemoteUDPAddr().Port)
+
+	err := gopacket.SerializeLayers(sender.ipBuf, sender.opts, sender.udpHeader, &payload)
+	if err != nil {
+		return err
+	}
+	packet := sender.ipBuf.Bytes()
+	_, err = sender.socket.Write(packet)
+	if err == nil || PosixError(err) != syscall.EMSGSIZE {
+		return err
+	}
+	f, err := sender.socket.File()
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+	fd := int(f.Fd())
+	log.Println("EMSGSIZE on send, expecting PMTU update (IP packet was",
+		len(packet), "bytes, payload was", len(msg), "bytes)")
+	pmtu, err := syscall.GetsockoptInt(fd, syscall.IPPROTO_IP, syscall.IP_MTU)
+	if err != nil {
+		return err
+	}
+	return MsgTooBigError{PMTU: pmtu}
+}
+
+func (sender *RawUDPSender) Shutdown() error {
+	defer func() { sender.socket = nil }()
+	return sender.socket.Close()
+}
+
+func dialIP(conn *LocalConnection) (*net.IPConn, error) {
+	ipLocalAddr, err := ipAddr(conn.TCPConn.LocalAddr())
+	if err != nil {
+		return nil, err
+	}
+	ipRemoteAddr, err := ipAddr(conn.TCPConn.RemoteAddr())
+	if err != nil {
+		return nil, err
+	}
+	ipSocket, err := net.DialIP("ip4:UDP", ipLocalAddr, ipRemoteAddr)
+	if err != nil {
+		return nil, err
+	}
+	f, err := ipSocket.File()
+	if err != nil {
+		return nil, err
+	}
+	defer f.Close()
+	fd := int(f.Fd())
+	// This Makes sure all packets we send out have DF set on them.
+	err = syscall.SetsockoptInt(fd, syscall.IPPROTO_IP, syscall.IP_MTU_DISCOVER, syscall.IP_PMTUDISC_DO)
+	if err != nil {
+		return nil, err
+	}
+	return ipSocket, nil
+}
+
+func ipAddr(addr net.Addr) (*net.IPAddr, error) {
+	host, _, err := net.SplitHostPort(addr.String())
+	if err != nil {
+		return nil, err
+	}
+	return &net.IPAddr{
+		IP:   net.ParseIP(host),
+		Zone: ""}, nil
+}

router/mac_cache.go

@@ -0,0 +1,111 @@
+package weave
+
+import (
+	"bytes"
+	"fmt"
+	"net"
+	"sync"
+	"time"
+)
+
+type MacCacheEntry struct {
+	lastSeen time.Time
+	peer     *Peer
+}
+
+type MacCache struct {
+	sync.RWMutex
+	table       map[uint64]*MacCacheEntry
+	maxAge      time.Duration
+	expiryTimer *time.Timer
+	onExpiry    func(net.HardwareAddr, *Peer)
+}
+
+func NewMacCache(maxAge time.Duration, onExpiry func(net.HardwareAddr, *Peer)) *MacCache {
+	res := &MacCache{
+		table:    make(map[uint64]*MacCacheEntry),
+		maxAge:   maxAge,
+		onExpiry: onExpiry}
+	res.setExpiryTimer()
+	return res
+}
+
+func (cache *MacCache) Enter(mac net.HardwareAddr, peer *Peer) bool {
+	key := macint(mac)
+	now := time.Now()
+	cache.RLock()
+	entry, found := cache.table[key]
+	if found && entry.peer == peer && now.Before(entry.lastSeen.Add(cache.maxAge/10)) {
+		cache.RUnlock()
+		return false
+	} else {
+		cache.RUnlock()
+	}
+	cache.Lock()
+	defer cache.Unlock()
+	entry, found = cache.table[key]
+	if !found {
+		cache.table[key] = &MacCacheEntry{lastSeen: now, peer: peer}
+		return true
+	}
+	if entry.peer != peer {
+		entry.lastSeen = now
+		entry.peer = peer
+		return true
+	}
+	if now.After(entry.lastSeen.Add(cache.maxAge / 10)) {
+		entry.lastSeen = now
+	}
+	return false
+}
+
+func (cache *MacCache) Lookup(mac net.HardwareAddr) (*Peer, bool) {
+	key := macint(mac)
+	cache.RLock()
+	defer cache.RUnlock()
+	entry, found := cache.table[key]
+	if !found {
+		return nil, false
+	}
+	return entry.peer, true
+}
+
+func (cache *MacCache) Delete(peer *Peer) bool {
+	found := false
+	cache.Lock()
+	defer cache.Unlock()
+	for key, entry := range cache.table {
+		if entry.peer == peer {
+			delete(cache.table, key)
+			found = true
+		}
+	}
+	return found
+}
+
+func (cache *MacCache) String() string {
+	var buf bytes.Buffer
+	cache.RLock()
+	defer cache.RUnlock()
+	for key, entry := range cache.table {
+		buf.WriteString(fmt.Sprintf("%v -> %s (%v)\n", intmac(key), entry.peer.Name, entry.lastSeen))
+	}
+	return buf.String()
+}
+
+func (cache *MacCache) setExpiryTimer() {
+	cache.expiryTimer = time.AfterFunc(cache.maxAge/10, func() { cache.expire() })
+}
+
+func (cache *MacCache) expire() {
+	now := time.Now()
+	cache.Lock()
+	defer cache.Unlock()
+	for key, entry := range cache.table {
+		if now.After(entry.lastSeen.Add(cache.maxAge)) {
+			delete(cache.table, key)
+			cache.onExpiry(intmac(key), entry.peer)
+		}
+	}
+	cache.setExpiryTimer()
+}

router/pcap.go

@@ -0,0 +1,74 @@
+package weave
+
+import (
+	"code.google.com/p/gopacket/pcap"
+)
+
+type PcapIO struct {
+	handle *pcap.Handle
+}
+
+func NewPcapIO(ifName string, bufSz int) (PacketSourceSink, error) {
+	pio, err := newPcapIO(ifName, true, 65535, bufSz)
+	if err != nil {
+		return pio, err
+	}
+
+	// Under Linux, libpcap implements the SetDirection filtering
+	// in userspace.  So set a BPF filter to discard outbound
+	// packets inside the kernel.  We do this here rather than in
+	// newPcapIO because libpcap doesn't like this in combination
+	// with a 0 snaplen.
+	err = pio.handle.SetBPFFilter("inbound")
+	return pio, err
+}
+
+func NewPcapO(ifName string) (po PacketSink, err error) {
+	po, err = newPcapIO(ifName, false, 0, 0)
+	return
+}
+
+func newPcapIO(ifName string, promisc bool, snaplen int, bufSz int) (handle *PcapIO, err error) {
+	inactive, err := pcap.NewInactiveHandle(ifName)
+	if err != nil {
+		return
+	}
+	defer inactive.CleanUp()
+	if err = inactive.SetPromisc(promisc); err != nil {
+		return
+	}
+	if err = inactive.SetSnapLen(snaplen); err != nil {
+		return
+	}
+	if err = inactive.SetTimeout(MaxDuration); err != nil {
+		return
+	}
+	if err = inactive.SetImmediateMode(true); err != nil {
+		return
+	}
+	if err = inactive.SetBufferSize(bufSz); err != nil {
+		return
+	}
+	active, err := inactive.Activate()
+	if err != nil {
+		return
+	}
+	if err = active.SetDirection(pcap.DirectionIn); err != nil {
+		return
+	}
+	return &PcapIO{handle: active}, nil
+}
+
+func (pi *PcapIO) ReadPacket() (data []byte, err error) {
+	for {
+		data, _, err = pi.handle.ZeroCopyReadPacketData()
+		if err == nil || err != pcap.NextErrorTimeoutExpired {
+			break
+		}
+	}
+	return
+}
+
+func (po *PcapIO) WritePacket(data []byte) error {
+	return po.handle.WritePacketData(data)
+}

router/peer.go

@@ -0,0 +1,377 @@
+package weave
+
+import (
+	"fmt"
+	"log"
+	"net"
+	"sort"
+)
+
+func NewPeer(name PeerName, uid uint64, version uint64, router *Router) *Peer {
+	if uid == 0 {
+		uid = randUint64()
+	}
+	return &Peer{
+		Name:        name,
+		NameByte:    name.Bin(),
+		connections: make(map[PeerName]Connection),
+		version:     version,
+		UID:         uid,
+		Router:      router}
+}
+
+func (peer *Peer) Version() uint64 {
+	peer.RLock()
+	defer peer.RUnlock()
+	return peer.version
+}
+
+func (peer *Peer) IncrementLocalRefCount() {
+	peer.Lock()
+	defer peer.Unlock()
+	peer.localRefCount += 1
+}
+
+func (peer *Peer) DecrementLocalRefCount() {
+	peer.Lock()
+	defer peer.Unlock()
+	peer.localRefCount -= 1
+}
+
+func (peer *Peer) IsLocallyReferenced() bool {
+	peer.RLock()
+	defer peer.RUnlock()
+	return peer.localRefCount != 0
+}
+
+func (peer *Peer) ConnectionCount() int {
+	peer.RLock()
+	defer peer.RUnlock()
+	return len(peer.connections)
+}
+
+func (peer *Peer) ConnectionTo(name PeerName) (Connection, bool) {
+	peer.RLock()
+	defer peer.RUnlock()
+	conn, found := peer.connections[name]
+	return conn, found // yes, you really can't inline that. FFS.
+}
+
+func (peer *Peer) ForEachConnection(fun func(PeerName, Connection)) {
+	peer.RLock()
+	defer peer.RUnlock()
+	for name, conn := range peer.connections {
+		fun(name, conn)
+	}
+}
+
+func (peer *Peer) HasPathTo(target *Peer, symmetric bool) (bool, map[PeerName]bool) {
+	peer.RLock()
+	defer peer.RUnlock()
+	reached := make(map[PeerName]bool)
+	nextWorklist := []*Peer{peer}
+	reached[peer.Name] = true
+	for len(nextWorklist) > 0 {
+		worklist := nextWorklist
+		sort.Sort(ListOfPeers(worklist))
+		nextWorklist = []*Peer{}
+		for _, curPeer := range worklist {
+			if curPeer == target {
+				return true, reached
+			}
+			for remoteName, conn := range curPeer.connections {
+				if _, found := reached[remoteName]; found {
+					continue
+				}
+				// By taking this lock, we ensure no changes can occur
+				// to any peer we have reached. As a result, if/when
+				// we reach the target, we know the entire path from
+				// peer to target is valid - no changes can have been
+				// made to that path during our calculation.
+				remote := conn.Remote()
+				remote.RLock()
+				defer remote.RUnlock()
+				if _, found := remote.connections[curPeer.Name]; !symmetric || found {
+					nextWorklist = append(nextWorklist, remote)
+					reached[remoteName] = true
+				}
+			}
+		}
+	}
+	return false, reached
+}
+
+func (peer *Peer) SetVersionAndConnections(version uint64, connections map[PeerName]Connection) {
+	if peer == peer.Router.Ourself {
+		log.Fatal("Attempt to set version and connections on the local peer", peer.Name)
+	}
+	peer.Lock()
+	defer peer.Unlock()
+	peer.version = version
+	peer.connections = connections
+}
+
+func (peer *Peer) Forward(dstPeer *Peer, df bool, frame []byte, dec *EthernetDecoder) error {
+	return peer.Relay(peer, dstPeer, df, frame, dec)
+}
+
+func (peer *Peer) Broadcast(df bool, frame []byte, dec *EthernetDecoder) error {
+	return peer.RelayBroadcast(peer, df, frame, dec)
+}
+
+func (peer *Peer) Relay(srcPeer, dstPeer *Peer, df bool, frame []byte, dec *EthernetDecoder) error {
+	relayPeerName, found := peer.Router.Topology.Unicast(dstPeer.Name)
+	if !found {
+		// Not necessarily an error as there could be a race with the
+		// dst disappearing whilst the frame is in flight
+		log.Println("Received packet for unknown destination:", dstPeer.Name)
+		return nil
+	}
+	conn, found := peer.ConnectionTo(relayPeerName)
+	if !found {
+		// Again, could just be a race, not necessarily an error
+		log.Println("Unable to find connection to relay peer", relayPeerName)
+		return nil
+	}
+	return conn.(*LocalConnection).Forward(df, &ForwardedFrame{
+		srcPeer: srcPeer,
+		dstPeer: dstPeer,
+		frame:   frame},
+		dec)
+}
+
+func (peer *Peer) RelayBroadcast(srcPeer *Peer, df bool, frame []byte, dec *EthernetDecoder) error {
+	nextHops := peer.Router.Topology.Broadcast(srcPeer.Name)
+	if len(nextHops) == 0 {
+		return nil
+	}
+	// We must not hold a read lock on peer during the conn.Forward
+	// below, since that is a potentially blocking operation (e.g. if
+	// the channel is full).
+	nextConns := make([]*LocalConnection, 0, len(nextHops))
+	peer.RLock()
+	for _, hopName := range nextHops {
+		conn, found := peer.connections[hopName]
+		// Again, !found could just be due to a race.
+		if found {
+			nextConns = append(nextConns, conn.(*LocalConnection))
+		}
+	}
+	peer.RUnlock()
+	var err error
+	for _, conn := range nextConns {
+		err = conn.Forward(df, &ForwardedFrame{
+			srcPeer: srcPeer,
+			dstPeer: conn.Remote(),
+			frame:   frame},
+			dec)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (peer *Peer) String() string {
+	peer.RLock()
+	defer peer.RUnlock()
+	return fmt.Sprint("Peer ", peer.Name, " (v", peer.version, ") (UID ", peer.UID, ")")
+}
+
+func (peer *Peer) StartLocalPeer() {
+	if peer.Router.Ourself != peer {
+		log.Fatal("Attempt to start peer which is not the local peer")
+	}
+	queryChan := make(chan *PeerInteraction, ChannelSize)
+	peer.queryChan = queryChan
+	go peer.queryLoop(queryChan)
+}
+
+func (peer *Peer) CreateConnection(peerAddr string, acceptNewPeer bool) error {
+	if err := peer.checkConnectionLimit(); err != nil {
+		return err
+	}
+	// We're dialing the remote so that means connections will come from random ports
+	addrStr := NormalisePeerAddr(peerAddr)
+	tcpAddr, tcpErr := net.ResolveTCPAddr("tcp4", addrStr)
+	udpAddr, udpErr := net.ResolveUDPAddr("udp4", addrStr)
+	if tcpErr != nil || udpErr != nil {
+		// they really should have the same value, but just in case...
+		if tcpErr == nil {
+			return udpErr
+		}
+		return tcpErr
+	}
+	tcpConn, err := net.DialTCP("tcp4", nil, tcpAddr)
+	if err != nil {
+		return err
+	}
+	connRemote := NewRemoteConnection(peer, nil, tcpConn.RemoteAddr().String())
+	NewLocalConnection(connRemote, acceptNewPeer, tcpConn, udpAddr, peer.Router)
+	return nil
+}
+
+// ACTOR client API
+
+const (
+	PAddConnection         = iota
+	PBroadcastTCP          = iota
+	PDeleteConnection      = iota
+	PConnectionEstablished = iota
+)
+
+// Async: rely on the peer to shut us down if we shouldn't be adding
+// ourselves, so therefore this can be async
+func (peer *Peer) AddConnection(conn *LocalConnection) {
+	peer.queryChan <- &PeerInteraction{
+		Interaction: Interaction{code: PAddConnection},
+		payload:     conn}
+}
+
+// Sync.
+func (peer *Peer) DeleteConnection(conn *LocalConnection) {
+	resultChan := make(chan interface{})
+	peer.queryChan <- &PeerInteraction{
+		Interaction: Interaction{code: PDeleteConnection, resultChan: resultChan},
+		payload:     conn}
+	<-resultChan
+}
+
+// Async.
+func (peer *Peer) ConnectionEstablished(conn *LocalConnection) {
+	peer.queryChan <- &PeerInteraction{
+		Interaction: Interaction{code: PConnectionEstablished},
+		payload:     conn}
+}
+
+// Async.
+func (peer *Peer) BroadcastTCP(msg []byte) {
+	peer.queryChan <- &PeerInteraction{
+		Interaction: Interaction{code: PBroadcastTCP},
+		payload:     msg}
+}
+
+// ACTOR server
+
+func (peer *Peer) queryLoop(queryChan <-chan *PeerInteraction) {
+	for {
+		query, ok := <-queryChan
+		if !ok {
+			return
+		}
+		switch query.code {
+		case PAddConnection:
+			peer.handleAddConnection(query.payload.(*LocalConnection))
+		case PDeleteConnection:
+			peer.handleDeleteConnection(query.payload.(*LocalConnection))
+			query.resultChan <- nil
+		case PConnectionEstablished:
+			peer.handleConnectionEstablished(query.payload.(*LocalConnection))
+		case PBroadcastTCP:
+			peer.handleBroadcastTCP(query.payload.([]byte))
+		}
+	}
+}
+
+func (peer *Peer) handleAddConnection(conn *LocalConnection) {
+	if peer != conn.Local() {
+		log.Fatal("Attempt made to add connection to peer where peer is not the source of connection")
+	}
+	if conn.Remote() == nil {
+		log.Fatal("Attempt made to add connection to peer with unknown remote peer")
+	}
+	toName := conn.Remote().Name
+	dupErr := fmt.Errorf("Multiple connections to %s added to %s", toName, peer.Name)
+	// deliberately non symmetrical
+	if dupConn, found := peer.connections[toName]; found {
+		if dupConn == conn {
+			return
+		} else {
+			// conn.UID is used as the tie breaker here, in the
+			// knowledge that both sides will make the same decision.
+			dupConnLocal := dupConn.(*LocalConnection)
+			if conn.UID == dupConnLocal.UID {
+				// oh good grief. Sod it, just kill both of them.
+				conn.CheckFatal(dupErr)
+				dupConnLocal.CheckFatal(dupErr)
+				peer.handleDeleteConnection(dupConnLocal)
+				return
+			} else if conn.UID < dupConnLocal.UID {
+				dupConnLocal.CheckFatal(dupErr)
+				peer.handleDeleteConnection(dupConnLocal)
+			} else {
+				conn.CheckFatal(dupErr)
+				return
+			}
+		}
+	}
+	if err := peer.checkConnectionLimit(); err != nil {
+		conn.CheckFatal(err)
+		return
+	}
+	peer.Lock()
+	peer.connections[toName] = conn
+	peer.Unlock()
+}
+
+func (peer *Peer) handleDeleteConnection(conn *LocalConnection) {
+	if peer != conn.Local() {
+		log.Fatal("Attempt made to delete connection from peer where peer is not the source of connection")
+	}
+	if conn.Remote() == nil {
+		log.Fatal("Attempt made to delete connection to peer with unknown remote peer")
+	}
+	toName := conn.Remote().Name
+	if connFound, found := peer.connections[toName]; !found || connFound != conn {
+		return
+	}
+	peer.Lock()
+	delete(peer.connections, toName)
+	peer.Unlock()
+	broadcast := false
+	if conn.Established() {
+		peer.Lock()
+		peer.version += 1
+		peer.Unlock()
+		broadcast = true
+	}
+	// Must do garbage collection first to ensure we don't send out an
+	// update with unreachable peers (can cause looping)
+	peer.Router.Peers.GarbageCollect(peer.Router)
+	if broadcast {
+		peer.broadcastPeerUpdate()
+	}
+}
+
+func (peer *Peer) handleConnectionEstablished(conn *LocalConnection) {
+	if peer != conn.Local() {
+		log.Fatal("Peer informed of active connection where peer is not the source of connection")
+	}
+	if dupConn, found := peer.connections[conn.Remote().Name]; !found || conn != dupConn {
+		conn.CheckFatal(fmt.Errorf("Cannot set unknown connection active"))
+		return
+	}
+	peer.Lock()
+	peer.version += 1
+	peer.Unlock()
+	log.Println("Peer", peer.Name, "established active connection to remote peer", conn.Remote().Name, "at", conn.RemoteTCPAddr())
+	peer.broadcastPeerUpdate(conn.Remote())
+}
+
+func (peer *Peer) handleBroadcastTCP(msg []byte) {
+	peer.Router.Topology.RebuildRoutes()
+	peer.ForEachConnection(func(_ PeerName, conn Connection) {
+		conn.(*LocalConnection).SendTCP(msg)
+	})
+}
+
+func (peer *Peer) broadcastPeerUpdate(peers ...*Peer) {
+	peer.handleBroadcastTCP(Concat(ProtocolUpdateByte, EncodePeers(append(peers, peer)...)))
+}
+
+func (peer *Peer) checkConnectionLimit() error {
+	if 0 != peer.Router.ConnLimit && peer.ConnectionCount() >= peer.Router.ConnLimit {
+		return fmt.Errorf("Connection limit reached (%v)", peer.Router.ConnLimit)
+	}
+	return nil
+}

router/peer_cache.go

@@ -0,0 +1,315 @@
+package weave
+
+import (
+	"bytes"
+	"encoding/gob"
+	"fmt"
+	"io"
+	"sync"
+)
+
+type PeerCache struct {
+	sync.RWMutex
+	table map[PeerName]*Peer
+	onGC  func(*Peer)
+}
+
+func NewPeerCache(onGC func(*Peer)) *PeerCache {
+	return &PeerCache{table: make(map[PeerName]*Peer), onGC: onGC}
+}
+
+func (cache *PeerCache) FetchWithDefault(peer *Peer) *Peer {
+	cache.RLock()
+	res, found := cache.fetchAlias(peer)
+	cache.RUnlock()
+	if found {
+		return res
+	}
+	cache.Lock()
+	defer cache.Unlock()
+	res, found = cache.fetchAlias(peer)
+	if found {
+		return res
+	}
+	cache.table[peer.Name] = peer
+	peer.IncrementLocalRefCount()
+	return peer
+}
+
+func (cache *PeerCache) Fetch(name PeerName) (*Peer, bool) {
+	cache.RLock()
+	defer cache.RUnlock()
+	peer, found := cache.table[name]
+	return peer, found // GRRR, why can't I inline this!?
+}
+
+func (cache *PeerCache) ForEach(fun func(PeerName, *Peer)) {
+	cache.RLock()
+	defer cache.RUnlock()
+	for name, peer := range cache.table {
+		fun(name, peer)
+	}
+}
+
+func (cache *PeerCache) ApplyUpdate(update []byte, router *Router) ([]byte, error) {
+	cache.Lock()
+
+	newPeers, decodedUpdate, decodedConns, err := cache.decodeUpdate(update, router)
+	if err != nil {
+		cache.Unlock()
+		return nil, err
+	}
+
+	// By this point, we know the update doesn't refer to any peers we
+	// have no knowledge of. We can now apply the update. Start by
+	// adding in any new peers into the cache.
+	for name, newPeer := range newPeers {
+		cache.table[name] = newPeer
+	}
+
+	// Now apply the updates
+	newUpdate := cache.applyUpdate(decodedUpdate, decodedConns, router)
+
+	for _, peerRemoved := range cache.garbageCollect(router.Ourself) {
+		delete(newUpdate, peerRemoved.Name)
+	}
+
+	// Don't need to hold cache lock any longer
+	cache.Unlock()
+
+	if len(newUpdate) > 0 {
+		router.ConnectionMaker.Refresh()
+	}
+
+	return encodePeersMap(newUpdate), nil
+}
+
+func (cache *PeerCache) EncodeAllPeers() []byte {
+	cache.RLock()
+	defer cache.RUnlock()
+	return encodePeersMap(cache.table)
+}
+
+func EncodePeers(peers ...*Peer) []byte {
+	buf := new(bytes.Buffer)
+	enc := gob.NewEncoder(buf)
+	for _, peer := range peers {
+		peer.encodePeer(enc)
+	}
+	return buf.Bytes()
+}
+
+func (cache *PeerCache) GarbageCollect(router *Router) []*Peer {
+	cache.Lock()
+	defer cache.Unlock()
+	return cache.garbageCollect(router.Ourself)
+}
+
+func (cache *PeerCache) String() string {
+	var buf bytes.Buffer
+	cache.ForEach(func(name PeerName, peer *Peer) {
+		buf.WriteString(fmt.Sprint(peer, "\n"))
+		peer.ForEachConnection(func(remoteName PeerName, conn Connection) {
+			buf.WriteString(fmt.Sprintf("   -> %v [%v]\n", remoteName, conn.RemoteTCPAddr()))
+		})
+	})
+	return buf.String()
+}
+
+func (cache *PeerCache) fetchAlias(peer *Peer) (*Peer, bool) {
+	if existingPeer, found := cache.table[peer.Name]; found {
+		if existingPeer.UID == peer.UID {
+			existingPeer.IncrementLocalRefCount()
+			return existingPeer, true
+		} else {
+			return nil, true
+		}
+	}
+	return nil, false
+}
+
+func (cache *PeerCache) garbageCollect(ourself *Peer) []*Peer {
+	removed := []*Peer{}
+	for name, peer := range cache.table {
+		found, _ := ourself.HasPathTo(peer, false)
+		if !found && !peer.IsLocallyReferenced() {
+			cache.onGC(peer)
+			delete(cache.table, name)
+			ourself.Router.Macs.Delete(peer)
+			removed = append(removed, peer)
+		}
+	}
+	return removed
+}
+
+func encodePeersMap(peers map[PeerName]*Peer) []byte {
+	buf := new(bytes.Buffer)
+	enc := gob.NewEncoder(buf)
+	for _, peer := range peers {
+		peer.encodePeer(enc)
+	}
+	return buf.Bytes()
+}
+
+func (cache *PeerCache) decodeUpdate(update []byte, router *Router) (newPeers map[PeerName]*Peer, decodedUpdate []*Peer, decodedConns [][]byte, err error) {
+	newPeers = make(map[PeerName]*Peer)
+	decodedUpdate = []*Peer{}
+	decodedConns = [][]byte{}
+
+	updateBuf := new(bytes.Buffer)
+	updateBuf.Write(update)
+	decoder := gob.NewDecoder(updateBuf)
+
+	for {
+		nameByte, uid, version, connsBuf, decErr := decodePeerNoConns(decoder)
+		if decErr == io.EOF {
+			break
+		} else if decErr != nil {
+			err = decErr
+			return
+		}
+		name := PeerNameFromBin(nameByte)
+		newPeer := NewPeer(name, uid, version, router)
+		decodedUpdate = append(decodedUpdate, newPeer)
+		decodedConns = append(decodedConns, connsBuf)
+		existingPeer, found := cache.table[name]
+		if !found {
+			newPeers[name] = newPeer
+		} else if existingPeer.UID != newPeer.UID {
+			err = NameCollisionError{Name: newPeer.Name}
+			return
+		}
+	}
+
+	unknownPeers := false
+	for _, connsBuf := range decodedConns {
+		connsIterator(connsBuf, func(remoteNameByte []byte, _ string) {
+			remoteName := PeerNameFromBin(remoteNameByte)
+			if _, found := newPeers[remoteName]; found {
+				return
+			}
+			if _, found := cache.table[remoteName]; found {
+				return
+			}
+			// Update refers to a peer which we have no knowledge
+			// of. Thus we can't apply the update. Abort.
+			unknownPeers = true
+		})
+	}
+	if unknownPeers {
+		err = UnknownPeersError{}
+		return
+	}
+	return
+}
+
+func (cache *PeerCache) applyUpdate(decodedUpdate []*Peer, decodedConns [][]byte, router *Router) map[PeerName]*Peer {
+	newUpdate := make(map[PeerName]*Peer)
+	for idx, newPeer := range decodedUpdate {
+		connsBuf := decodedConns[idx]
+		name := newPeer.Name
+		// guaranteed to find peer in the cache.table
+		peer := cache.table[name]
+		if peer != newPeer {
+			if peer.Version() > newPeer.Version() {
+				// we know more about this one than the update. If
+				// peer is ourself, this is slightly racey (further
+				// changes could occur to ourself in the mean
+				// time). But it doesn't matter as we know that we're
+				// already > newPeer.Version() and that's not going to
+				// change.
+				newUpdate[name] = peer
+				continue
+			} else if peer == router.Ourself {
+				// nobody but us updates us
+				continue
+			} else if peer.Version() == newPeer.Version() {
+				// implication is that connections are equal too
+				continue
+			}
+		}
+		// If we're here, either it was a new peer, or the update has
+		// more info about the peer than we do. Either case, we need
+		// to set version and conns and include the updated peer in
+		// the outgoing update.
+
+		// Can peer have been updated by anyone else in the mean time?
+		// No - we know that peer is not ourself, so the only prospect
+		// for an update would be someone else calling
+		// router.Peers.ApplyUpdate. But ApplyUpdate takes the Lock on
+		// the peers cache, so there can be no race here.
+		conns := readConnsMap(peer, connsBuf, cache.table)
+		peer.SetVersionAndConnections(newPeer.Version(), conns)
+		newUpdate[name] = peer
+	}
+	return newUpdate
+}
+
+func (peer *Peer) encodePeer(enc *gob.Encoder) {
+	peer.RLock()
+	defer peer.RUnlock()
+
+	checkFatal(enc.Encode(peer.NameByte))
+	checkFatal(enc.Encode(peer.UID))
+	checkFatal(enc.Encode(peer.version))
+
+	connsBuf := new(bytes.Buffer)
+	connsEnc := gob.NewEncoder(connsBuf)
+	for _, conn := range peer.connections {
+		// DANGER holding rlock on peer, going to take rlock on conn
+		if !conn.Established() {
+			continue
+		}
+		checkFatal(connsEnc.Encode(conn.Remote().NameByte))
+		checkFatal(connsEnc.Encode(conn.RemoteTCPAddr()))
+	}
+	checkFatal(enc.Encode(connsBuf.Bytes()))
+}
+
+func decodePeerNoConns(dec *gob.Decoder) (nameByte []byte, uid uint64, version uint64, conns []byte, err error) {
+	err = dec.Decode(&nameByte)
+	if err != nil {
+		return
+	}
+	err = dec.Decode(&uid)
+	if err != nil {
+		return
+	}
+	err = dec.Decode(&version)
+	if err != nil {
+		return
+	}
+	err = dec.Decode(&conns)
+	if err == io.EOF {
+		err = nil
+	}
+	return
+}
+
+func connsIterator(input []byte, fun func([]byte, string)) {
+	buf := new(bytes.Buffer)
+	buf.Write(input)
+	dec := gob.NewDecoder(buf)
+	for {
+		var nameByte []byte
+		err := dec.Decode(&nameByte)
+		if err == io.EOF {
+			return
+		}
+		checkFatal(err)
+		var foundAt string
+		checkFatal(dec.Decode(&foundAt))
+		fun(nameByte, string(foundAt))
+	}
+}
+
+func readConnsMap(peer *Peer, buf []byte, table map[PeerName]*Peer) map[PeerName]Connection {
+	conns := make(map[PeerName]Connection)
+	connsIterator(buf, func(nameByte []byte, remoteTCPAddr string) {
+		name := PeerNameFromBin(nameByte)
+		remotePeer := table[name]
+		conn := NewRemoteConnection(peer, remotePeer, remoteTCPAddr)
+		conns[name] = conn
+	})
+	return conns
+}

router/peer_name_hash.go

@@ -0,0 +1,45 @@
+// +build peer_name_hash
+
+// Let peer names be sha256 hashes...of anything (as long as it's
+// unique).
+package weave
+
+import (
+	"crypto/sha256"
+	"encoding/hex"
+)
+
+type PeerName string
+
+const (
+	PeerNameFlavour = "hash"
+	NameSize        = sha256.Size >> 1
+	UnknownPeerName = PeerName("")
+)
+
+func PeerNameFromUserInput(userInput string) (PeerName, error) {
+	// fixed-length identity
+	nameByteAry := sha256.Sum256([]byte(userInput))
+	return PeerNameFromBin(nameByteAry[:NameSize]), nil
+}
+
+func PeerNameFromString(nameStr string) (PeerName, error) {
+	if _, err := hex.DecodeString(nameStr); err != nil {
+		return UnknownPeerName, err
+	}
+	return PeerName(nameStr), nil
+}
+
+func PeerNameFromBin(nameByte []byte) PeerName {
+	return PeerName(hex.EncodeToString(nameByte))
+}
+
+func (name PeerName) Bin() []byte {
+	res, err := hex.DecodeString(string(name))
+	checkFatal(err)
+	return res
+}
+
+func (name PeerName) String() string {
+	return string(name)
+}

router/peer_name_mac.go

@@ -0,0 +1,57 @@
+// The !peer_name_alternative effectively makes this the default,
+// i.e. to choose an alternative, run "go build -tags
+// 'peer_name_alternative peer_name_hash'"
+
+// +build peer_name_mac !peer_name_alternative
+
+// Let peer names be MACs...
+//
+// MACs need to be unique across our network, or bad things will
+// happen anyway. So they make pretty good candidates for peer
+// names. And doing so is pretty efficient both computationally and
+// network overhead wise.
+//
+// Note that we do not mandate *what* MAC should be used as the peer
+// name. In particular it doesn't actually have to be the MAC of, say,
+// the network interface the peer is sniffing on.
+package weave
+
+import (
+	"net"
+)
+
+// PeerName is used as a map key. Since net.HardwareAddr isn't
+// suitable for that - it's a slice, and slices can't be map keys - we
+// convert that to/from uint64.
+type PeerName uint64
+
+const (
+	PeerNameFlavour = "mac"
+	NameSize        = 6
+	UnknownPeerName = PeerName(0)
+)
+
+func PeerNameFromUserInput(userInput string) (PeerName, error) {
+	res, err := PeerNameFromString(userInput)
+	return res, err
+}
+
+func PeerNameFromString(nameStr string) (PeerName, error) {
+	mac, err := net.ParseMAC(nameStr)
+	if err != nil {
+		return UnknownPeerName, err
+	}
+	return PeerName(macint(mac)), nil
+}
+
+func PeerNameFromBin(nameByte []byte) PeerName {
+	return PeerName(macint(net.HardwareAddr(nameByte)))
+}
+
+func (name PeerName) Bin() []byte {
+	return intmac(uint64(name))
+}
+
+func (name PeerName) String() string {
+	return intmac(uint64(name)).String()
+}

router/router.go

@@ -0,0 +1,291 @@
+package weave
+
+import (
+	"bytes"
+	"code.google.com/p/gopacket/layers"
+	"encoding/binary"
+	"fmt"
+	"io"
+	"log"
+	"net"
+	"syscall"
+	"time"
+)
+
+const macMaxAge = 10 * time.Minute
+
+func NewRouter(iface *net.Interface, name PeerName, password []byte, connLimit int, bufSz int, logFrame func(string, []byte, *layers.Ethernet)) *Router {
+	onMacExpiry := func(mac net.HardwareAddr, peer *Peer) {
+		log.Println("Expired MAC", mac, "at", peer.Name)
+	}
+	onPeerGC := func(peer *Peer) {
+		log.Println("Removing unreachable", peer)
+	}
+	router := &Router{
+		Iface:     iface,
+		Macs:      NewMacCache(macMaxAge, onMacExpiry),
+		Peers:     NewPeerCache(onPeerGC),
+		ConnLimit: connLimit,
+		BufSz:     bufSz,
+		LogFrame:  logFrame}
+	if len(password) > 0 {
+		router.Password = &password
+	}
+	ourself := NewPeer(name, 0, 0, router)
+	router.Ourself = router.Peers.FetchWithDefault(ourself)
+	router.Ourself.StartLocalPeer()
+	log.Println("Local identity is", router.Ourself.Name)
+
+	return router
+}
+
+func (router *Router) UsingPassword() bool {
+	return router.Password != nil
+}
+
+func (router *Router) Start() {
+	// we need two pcap handles since they aren't thread-safe
+	pio, err := NewPcapIO(router.Iface.Name, router.BufSz)
+	checkFatal(err)
+	po, err := NewPcapO(router.Iface.Name)
+	checkFatal(err)
+	router.ConnectionMaker = StartConnectionMaker(router)
+	router.Topology = StartTopology(router)
+	router.UDPListener = router.listenUDP(Port, po)
+	router.listenTCP(Port)
+	router.sniff(pio)
+}
+
+func (router *Router) Status() string {
+	var buf bytes.Buffer
+	buf.WriteString(fmt.Sprintln("Local identity is", router.Ourself.Name))
+	buf.WriteString(fmt.Sprintln("Sniffing traffic on", router.Iface))
+	buf.WriteString(fmt.Sprintf("MACs:\n%s", router.Macs))
+	buf.WriteString(fmt.Sprintf("Peers:\n%s", router.Peers))
+	buf.WriteString(fmt.Sprintf("Topology:\n%s", router.Topology))
+	buf.WriteString(fmt.Sprintf("Reconnects:\n%s", router.ConnectionMaker))
+	return buf.String()
+}
+
+func (router *Router) sniff(pio PacketSourceSink) {
+	log.Println("Sniffing traffic on", router.Iface)
+
+	dec := NewEthernetDecoder()
+	injectFrame := func(frame []byte) error { return pio.WritePacket(frame) }
+	checkFrameTooBig := func(err error) error { return dec.CheckFrameTooBig(err, injectFrame) }
+	mac := router.Iface.HardwareAddr
+	if router.Macs.Enter(mac, router.Ourself) {
+		log.Println("Discovered our MAC", mac)
+	}
+	go func() {
+		for {
+			pkt, err := pio.ReadPacket()
+			checkFatal(err)
+			router.LogFrame("Sniffed", pkt, nil)
+			checkWarn(router.handleCapturedPacket(pkt, dec, checkFrameTooBig))
+		}
+	}()
+}
+
+func (router *Router) handleCapturedPacket(frameData []byte, dec *EthernetDecoder, checkFrameTooBig func(error) error) error {
+	dec.DecodeLayers(frameData)
+	decodedLen := len(dec.decoded)
+	if decodedLen == 0 {
+		return nil
+	}
+	srcMac := dec.eth.SrcMAC
+	srcPeer, found := router.Macs.Lookup(srcMac)
+	// We need to filter out frames we injected ourselves. For such
+	// frames, the srcMAC will have been recorded as associated with a
+	// different peer.
+	if found && srcPeer != router.Ourself {
+		return nil
+	}
+	if router.Macs.Enter(srcMac, router.Ourself) {
+		log.Println("Discovered local MAC", srcMac)
+	}
+	if dec.DropFrame() {
+		return nil
+	}
+	dstMac := dec.eth.DstMAC
+	dstPeer, found := router.Macs.Lookup(dstMac)
+	if found && dstPeer == router.Ourself {
+		return nil
+	}
+	df := decodedLen == 2 && (dec.ip.Flags&layers.IPv4DontFragment != 0)
+	if df {
+		router.LogFrame("Forwarding DF", frameData, &dec.eth)
+	} else {
+		router.LogFrame("Forwarding", frameData, &dec.eth)
+	}
+	// at this point we are handing over the frame to forwarders, so
+	// we need to make a copy of it in order to prevent the next
+	// capture from overwriting the data
+	frameLen := len(frameData)
+	frameCopy := make([]byte, frameLen, frameLen)
+	copy(frameCopy, frameData)
+	if !found || dec.BroadcastFrame() {
+		return checkFrameTooBig(router.Ourself.Broadcast(df, frameCopy, dec))
+	} else {
+		return checkFrameTooBig(router.Ourself.Forward(dstPeer, df, frameCopy, dec))
+	}
+}
+
+func (router *Router) listenTCP(localPort int) {
+	localAddr, err := net.ResolveTCPAddr("tcp4", fmt.Sprint(":", localPort))
+	checkFatal(err)
+	ln, err := net.ListenTCP("tcp4", localAddr)
+	checkFatal(err)
+	go func() {
+		defer ln.Close()
+		for {
+			tcpConn, err := ln.AcceptTCP()
+			if err != nil {
+				log.Println(err)
+				continue
+			}
+			router.acceptTCP(tcpConn)
+		}
+	}()
+}
+
+func (router *Router) acceptTCP(tcpConn *net.TCPConn) {
+	// someone else is dialing us, so our udp sender is the conn
+	// on Port and we wait for them to send us something on UDP to
+	// start.
+	connRemote := NewRemoteConnection(router.Ourself, nil, tcpConn.RemoteAddr().String())
+	NewLocalConnection(connRemote, true, tcpConn, nil, router)
+}
+
+func (router *Router) listenUDP(localPort int, po PacketSink) *net.UDPConn {
+	localAddr, err := net.ResolveUDPAddr("udp4", fmt.Sprint(":", localPort))
+	checkFatal(err)
+	conn, err := net.ListenUDP("udp4", localAddr)
+	checkFatal(err)
+	f, err := conn.File()
+	defer f.Close()
+	checkFatal(err)
+	fd := int(f.Fd())
+	// This one makes sure all packets we send out do not have DF set on them.
+	err = syscall.SetsockoptInt(fd, syscall.IPPROTO_IP, syscall.IP_MTU_DISCOVER, syscall.IP_PMTUDISC_DONT)
+	checkFatal(err)
+	go router.udpReader(conn, po)
+	return conn
+}
+
+func (router *Router) udpReader(conn *net.UDPConn, po PacketSink) {
+	defer conn.Close()
+	dec := NewEthernetDecoder()
+	handleUDPPacket := router.handleUDPPacketFunc(dec, po)
+	buf := make([]byte, MaxUDPPacketSize)
+	for {
+		n, sender, err := conn.ReadFromUDP(buf)
+		if err == io.EOF {
+			return
+		} else if err != nil {
+			log.Println("ignoring UDP read error", err)
+			continue
+		} else if n < NameSize {
+			continue // TODO something different?
+		} else {
+			name := PeerNameFromBin(buf[:NameSize])
+			packet := make([]byte, n-NameSize)
+			copy(packet, buf[NameSize:n])
+			udpPacket := &UDPPacket{
+				Name:   name,
+				Packet: packet,
+				Sender: sender}
+			peerConn, found := router.Ourself.ConnectionTo(name)
+			if !found {
+				continue
+			}
+			relayConn, ok := peerConn.(*LocalConnection)
+			if !ok {
+				continue
+			}
+			checkWarn(relayConn.Decryptor.IterateFrames(handleUDPPacket, udpPacket))
+		}
+	}
+}
+
+func (router *Router) handleUDPPacketFunc(dec *EthernetDecoder, po PacketSink) FrameConsumer {
+	checkFrameTooBig := func(err error, srcPeer *Peer) error {
+		if err == nil { // optimisation: avoid closure creation in common case
+			return nil
+		}
+		return dec.CheckFrameTooBig(err,
+			func(icmpFrame []byte) error {
+				return router.Ourself.Forward(srcPeer, false, icmpFrame, nil)
+			})
+	}
+
+	return func(relayConn *LocalConnection, sender *net.UDPAddr, srcNameByte, dstNameByte []byte, frameLen uint16, frame []byte) error {
+		srcName := PeerNameFromBin(srcNameByte)
+		dstName := PeerNameFromBin(dstNameByte)
+		srcPeer, found := router.Peers.Fetch(srcName)
+		if !found {
+			return nil
+		}
+		dstPeer, found := router.Peers.Fetch(dstName)
+		if !found {
+			return nil
+		}
+
+		dec.DecodeLayers(frame)
+		decodedLen := len(dec.decoded)
+		df := decodedLen == 2 && (dec.ip.Flags&layers.IPv4DontFragment != 0)
+		srcMac := dec.eth.SrcMAC
+
+		if dstPeer != router.Ourself {
+			// it's not for us, we're just relaying it
+			if decodedLen == 0 {
+				return nil
+			}
+			if router.Macs.Enter(srcMac, srcPeer) {
+				log.Println("Discovered remote MAC", srcMac, "at", srcPeer.Name)
+			}
+			if df {
+				router.LogFrame("Relaying DF", frame, &dec.eth)
+			} else {
+				router.LogFrame("Relaying", frame, &dec.eth)
+			}
+			return checkFrameTooBig(router.Ourself.Relay(srcPeer, dstPeer, df, frame, dec), srcPeer)
+		}
+
+		if relayConn.Remote().Name == srcPeer.Name {
+			if frameLen == 0 {
+				relayConn.SetRemoteUDPAddr(sender)
+				return nil
+			} else if frameLen == FragTestSize && bytes.Equal(frame, FragTest) {
+				relayConn.SendTCP(ProtocolFragmentationReceivedByte)
+				return nil
+			} else if frameLen == PMTUDiscoverySize && bytes.Equal(frame, PMTUDiscovery) {
+				return nil
+			}
+		}
+
+		if decodedLen == 0 {
+			return nil
+		}
+
+		if dec.IsPMTUVerify() && relayConn.Remote().Name == srcPeer.Name {
+			frameLenBytes := []byte{0, 0}
+			binary.BigEndian.PutUint16(frameLenBytes, uint16(frameLen-EthernetOverhead))
+			relayConn.SendTCP(Concat(ProtocolPMTUVerifiedByte, frameLenBytes))
+			return nil
+		}
+
+		if router.Macs.Enter(srcMac, srcPeer) {
+			log.Println("Discovered remote MAC", srcMac, "at", srcPeer.Name)
+		}
+		router.LogFrame("Injecting", frame, &dec.eth)
+		checkWarn(po.WritePacket(frame))
+
+		dstPeer, found = router.Macs.Lookup(dec.eth.DstMAC)
+		if !found || dec.BroadcastFrame() || dstPeer != router.Ourself {
+			return checkFrameTooBig(router.Ourself.RelayBroadcast(srcPeer, df, frame, dec), srcPeer)
+		}
+
+		return nil
+	}
+}

router/topology.go

@@ -0,0 +1,175 @@
+package weave
+
+import (
+	"bytes"
+	"fmt"
+	"log"
+	"sort"
+)
+
+func (topo *Topology) Unicast(name PeerName) (PeerName, bool) {
+	topo.RLock()
+	defer topo.RUnlock()
+	hop, found := topo.unicast[name]
+	return hop, found
+}
+
+func (topo *Topology) Broadcast(name PeerName) []PeerName {
+	topo.RLock()
+	defer topo.RUnlock()
+	hops, found := topo.broadcast[name]
+	if !found {
+		return []PeerName{}
+	}
+	return hops
+}
+
+func (topo *Topology) String() string {
+	var buf bytes.Buffer
+	topo.RLock()
+	defer topo.RUnlock()
+	buf.WriteString(fmt.Sprintln("unicast:"))
+	for name, hop := range topo.unicast {
+		buf.WriteString(fmt.Sprintf("%s -> %s\n", name, hop))
+	}
+	buf.WriteString(fmt.Sprintln("broadcast:"))
+	for name, hops := range topo.broadcast {
+		buf.WriteString(fmt.Sprintf("%s -> %v\n", name, hops))
+	}
+	return buf.String()
+}
+
+func StartTopology(router *Router) *Topology {
+	queryChan := make(chan *Interaction, ChannelSize)
+	state := &Topology{
+		router:    router,
+		queryChan: queryChan,
+		unicast:   make(map[PeerName]PeerName),
+		broadcast: make(map[PeerName][]PeerName)}
+	state.unicast[router.Ourself.Name] = UnknownPeerName
+	state.broadcast[router.Ourself.Name] = []PeerName{}
+	go state.queryLoop(queryChan)
+	return state
+}
+
+// ACTOR client API
+
+const (
+	TFetchAll      = iota
+	TRebuildRoutes = iota
+)
+
+func (topo *Topology) FetchAll() []byte {
+	resultChan := make(chan interface{}, 0)
+	topo.queryChan <- &Interaction{
+		code:       TFetchAll,
+		resultChan: resultChan}
+	result := <-resultChan
+	return result.([]byte)
+}
+
+// Async.
+func (topo *Topology) RebuildRoutes() {
+	topo.queryChan <- &Interaction{code: TRebuildRoutes}
+}
+
+// ACTOR server
+
+func (topo *Topology) queryLoop(queryChan <-chan *Interaction) {
+	for {
+		query, ok := <-queryChan
+		if !ok {
+			return
+		}
+		switch query.code {
+		case TRebuildRoutes:
+			unicast := topo.buildUnicastRoutes()
+			broadcast := topo.buildBroadcastRoutes()
+			topo.Lock()
+			topo.unicast = unicast
+			topo.broadcast = broadcast
+			topo.Unlock()
+		case TFetchAll:
+			query.resultChan <- topo.router.Peers.EncodeAllPeers()
+		default:
+			log.Fatal("Unexpected topology query:", query)
+		}
+	}
+}
+
+func (topo *Topology) buildUnicastRoutes() map[PeerName]PeerName {
+	// He we are calculating all the routes for the question: if *we*
+	// want to send a packet to Peer X, what is the next hop?
+	//
+	// When we sniff a packet, we determine the destination peer
+	// ourself. Consequently, we can relay the packet via any
+	// arbitrary peers - the intermediate peers do not have to have
+	// any knowledge of the MAC address at all. Thus there's no need
+	// to exchange knowledge of MAC addresses, nor any constraints on
+	// the routes that we construct.
+	//
+	// Because currently we do not have weightings on the connections
+	// between peers, there is no need to use Dijkstra's
+	// algorithm. Instead, we can use the simpler and cheaper
+	// breadth-first widening.
+	//
+	// Currently we make no attempt to perform any balancing of any
+	// sort. That could happen in the future.
+
+	unicast := make(map[PeerName]PeerName)
+	ourself := topo.router.Ourself
+	unicast[ourself.Name] = UnknownPeerName
+	nextWorklist := []*Peer{}
+	ourself.ForEachConnection(func(remoteName PeerName, conn Connection) {
+		unicast[remoteName] = remoteName
+		nextWorklist = append(nextWorklist, conn.Remote())
+	})
+	for len(nextWorklist) > 0 {
+		worklist := nextWorklist
+		sort.Sort(ListOfPeers(worklist))
+		nextWorklist = []*Peer{}
+		for _, peer := range worklist {
+			localName := peer.Name
+			peer.ForEachConnection(func(remoteName PeerName, conn Connection) {
+				if _, found := unicast[remoteName]; found {
+					return
+				}
+				// We now know how to get to remoteName: the same
+				// way we get to peer. However, we only record this as
+				// a valid path if remoteName has peer in its
+				// connections. I.e. it must be a symmetric
+				// connection.
+				if _, found := conn.Remote().ConnectionTo(localName); found {
+					unicast[remoteName] = unicast[localName]
+					nextWorklist = append(nextWorklist, conn.Remote())
+				}
+			})
+		}
+	}
+	return unicast
+}
+
+func (topo *Topology) buildBroadcastRoutes() map[PeerName][]PeerName {
+	// For broadcast, we need to construct which of our connected
+	// peers we should pass frames on to should we receive any
+	// broadcast originally from peer X
+	broadcast := make(map[PeerName][]PeerName)
+	ourself := topo.router.Ourself
+
+	topo.router.Peers.ForEach(func(name PeerName, peer *Peer) {
+		hops := []PeerName{}
+		found, reached := peer.HasPathTo(ourself, true)
+		if found {
+			ourself.ForEachConnection(func(remoteName PeerName, conn Connection) {
+				if _, found := reached[remoteName]; found {
+					return
+				}
+				if _, found := conn.Remote().ConnectionTo(ourself.Name); found {
+					hops = append(hops, remoteName)
+				}
+			})
+		}
+		broadcast[name] = hops
+	})
+	return broadcast
+}

router/types.go

@@ -0,0 +1,309 @@
+package weave
+
+import (
+	"bytes"
+	"code.google.com/p/go-bit/bit"
+	"code.google.com/p/gopacket"
+	"code.google.com/p/gopacket/layers"
+	"encoding/gob"
+	"net"
+	"sync"
+	"time"
+)
+
+type Router struct {
+	Ourself            *Peer
+	Iface              *net.Interface
+	PeersSubscribeChan chan<- chan<- map[string]Peer
+	Macs               *MacCache
+	Peers              *PeerCache
+	UDPListener        *net.UDPConn
+	Topology           *Topology
+	ConnectionMaker    *ConnectionMaker
+	Password           *[]byte
+	ConnLimit          int
+	BufSz              int
+	LogFrame           func(string, []byte, *layers.Ethernet)
+}
+
+type Peer struct {
+	sync.RWMutex
+	Name          PeerName
+	NameByte      []byte
+	connections   map[PeerName]Connection
+	version       uint64
+	UID           uint64
+	Router        *Router
+	localRefCount uint64
+	queryChan     chan<- *PeerInteraction
+}
+
+type PeerInteraction struct {
+	Interaction
+	payload interface{}
+}
+
+type Connection interface {
+	Local() *Peer
+	Remote() *Peer
+	RemoteTCPAddr() string
+	Established() bool
+	Shutdown()
+}
+
+type RemoteConnection struct {
+	local         *Peer
+	remote        *Peer
+	remoteTCPAddr string
+}
+
+type LocalConnection struct {
+	sync.RWMutex
+	RemoteConnection
+	TCPConn       *net.TCPConn
+	tcpSender     TCPSender
+	remoteUDPAddr *net.UDPAddr
+	established   bool
+	stackFrag     bool
+	effectivePMTU int
+	SessionKey    *[32]byte
+	heartbeatStop chan<- interface{}
+	forwardChan   chan<- *ForwardedFrame
+	forwardChanDF chan<- *ForwardedFrame
+	stopForward   chan<- interface{}
+	stopForwardDF chan<- interface{}
+	verifyPMTU    chan<- int
+	Decryptor     Decryptor
+	Router        *Router
+	UID           uint64
+	shutdown      bool
+	queryChan     chan<- *ConnectionInteraction
+}
+
+type ConnectionInteraction struct {
+	Interaction
+	payload interface{}
+}
+
+type Forwarder struct {
+	conn            *LocalConnection
+	ch              <-chan *ForwardedFrame
+	stop            <-chan interface{}
+	verifyPMTUTick  <-chan time.Time
+	verifyPMTU      <-chan int
+	pmtuVerifyCount uint
+	enc             Encryptor
+	udpSender       UDPSender
+	maxPayload      int
+	pmtuVerified    bool
+	highestGoodPMTU int
+	unverifiedPMTU  int
+	lowestBadPMTU   int
+}
+
+type UDPPacket struct {
+	Name   PeerName
+	Packet []byte
+	Sender *net.UDPAddr
+}
+
+type EthernetDecoder struct {
+	eth     layers.Ethernet
+	ip      layers.IPv4
+	decoded []gopacket.LayerType
+	parser  *gopacket.DecodingLayerParser
+}
+
+type MsgTooBigError struct {
+	PMTU int // actual pmtu, i.e. what the kernel told us
+}
+
+type FrameTooBigError struct {
+	EPMTU int // effective pmtu, i.e. what we tell packet senders
+}
+
+type UnknownPeersError struct {
+}
+
+type NameCollisionError struct {
+	Name PeerName
+}
+
+type LocalAddress struct {
+	ip      net.IP
+	network *net.IPNet
+}
+
+type ForwardedFrame struct {
+	srcPeer *Peer
+	dstPeer *Peer
+	frame   []byte
+}
+
+type Interaction struct {
+	code       int
+	resultChan chan<- interface{}
+}
+
+type Topology struct {
+	sync.RWMutex
+	queryChan chan<- *Interaction
+	unicast   map[PeerName]PeerName
+	broadcast map[PeerName][]PeerName
+	router    *Router
+}
+
+type ConnectionMakerInteraction struct {
+	Interaction
+	address string
+}
+
+type ConnectionMaker struct {
+	router         *Router
+	queryChan      chan<- *ConnectionMakerInteraction
+	targets        map[string]*Target
+	cmdLineAddress map[string]bool
+}
+
+type ConnectionState int
+
+// Information about an address where we may find a peer
+type Target struct {
+	attempting   bool          // are we currently attempting to connect there?
+	tryAfter     time.Time     // next time to try this address
+	tryInterval  time.Duration // backoff time on next failure
+}
+
+// UDPSender interface and implementations
+
+type UDPSender interface {
+	Send([]byte) error
+	Shutdown() error
+}
+
+type SimpleUDPSender struct {
+	conn    *LocalConnection
+	udpConn *net.UDPConn
+}
+
+type RawUDPSender struct {
+	ipBuf     gopacket.SerializeBuffer
+	opts      gopacket.SerializeOptions
+	udpHeader *layers.UDP
+	socket    *net.IPConn
+	conn      *LocalConnection
+}
+
+// TCPSender interface and implementations
+
+type TCPSender interface {
+	Send([]byte) error
+}
+
+type SimpleTCPSender struct {
+	encoder *gob.Encoder
+}
+
+type EncryptedTCPSender struct {
+	sync.RWMutex
+	outerEncoder *gob.Encoder
+	innerEncoder *gob.Encoder
+	buffer       *bytes.Buffer
+	conn         *LocalConnection
+	msgCount     int
+}
+
+type EncryptedTCPMessage struct {
+	Number int
+	Body   []byte
+}
+
+// TCPReceiver interface and implementations
+
+type TCPReceiver interface {
+	Decode([]byte) ([]byte, error)
+}
+
+type SimpleTCPReceiver struct {
+}
+
+type EncryptedTCPReceiver struct {
+	conn     *LocalConnection
+	decoder  *gob.Decoder
+	buffer   *bytes.Buffer
+	msgCount int
+}
+
+// Encryptor interface and implementations
+
+type Encryptor interface {
+	FrameOverhead() int
+	PacketOverhead() int
+	IsEmpty() bool
+	Bytes() []byte
+	AppendFrame(*ForwardedFrame)
+	TotalLen() int
+}
+
+type NonEncryptor struct {
+	buf       []byte
+	bufTail   []byte
+	buffered  int
+	prefixLen int
+}
+
+type NaClEncryptor struct {
+	NonEncryptor
+	buf       []byte
+	offset    uint16
+	nonce     *[24]byte
+	nonceChan chan *[24]byte
+	flags     uint16
+	prefixLen int
+	conn      *LocalConnection
+	df        bool
+}
+
+// Decryptor interface and implementations
+
+type FrameConsumer func(*LocalConnection, *net.UDPAddr, []byte, []byte, uint16, []byte) error
+
+type Decryptor interface {
+	IterateFrames(FrameConsumer, *UDPPacket) error
+	ReceiveNonce([]byte)
+	Shutdown()
+}
+
+type NonDecryptor struct {
+	conn *LocalConnection
+}
+
+type NaClDecryptor struct {
+	NonDecryptor
+	instance   *NaClDecryptorInstance
+	instanceDF *NaClDecryptorInstance
+}
+
+type NaClDecryptorInstance struct {
+	nonce               *[24]byte
+	previousNonce       *[24]byte
+	usedOffsets         *bit.Set
+	previousUsedOffsets *bit.Set
+	highestOffsetSeen   uint16
+	nonceChan           chan *[24]byte
+}
+
+// Packet capture/inject interfaces
+
+type PacketSource interface {
+	ReadPacket() ([]byte, error)
+}
+
+type PacketSink interface {
+	WritePacket([]byte) error
+}
+
+type PacketSourceSink interface {
+	PacketSource
+	PacketSink
+}

router/utils.go

@@ -0,0 +1,110 @@
+package weave
+
+import (
+	"crypto/rand"
+	"fmt"
+	"log"
+	"net"
+)
+
+func checkFatal(e error) {
+	if e != nil {
+		log.Fatal(e)
+	}
+}
+
+func checkWarn(e error) {
+	if e != nil {
+		log.Println(e)
+	}
+}
+
+func PosixError(err error) error {
+	if err == nil {
+		return nil
+	}
+	operr, ok := err.(*net.OpError)
+	if !ok {
+		return nil
+	}
+	return operr.Err
+}
+
+func (mtbe MsgTooBigError) Error() string {
+	return fmt.Sprint("Msg too big error. PMTU is ", mtbe.PMTU)
+}
+
+func (ftbe FrameTooBigError) Error() string {
+	return fmt.Sprint("Frame too big error. Effective PMTU is ", ftbe.EPMTU)
+}
+
+func (upe UnknownPeersError) Error() string {
+	return fmt.Sprint("Reference to unknown peers")
+}
+
+func (nce NameCollisionError) Error() string {
+	return fmt.Sprint("Multiple peers found with same name: ", nce.Name)
+}
+
+func (packet UDPPacket) String() string {
+	return fmt.Sprintf("UDP Packet\n name: %s\n sender: %v\n payload: % X", packet.Name, packet.Sender, packet.Packet)
+}
+
+func Concat(elems ...[]byte) []byte {
+	res := []byte{}
+	for _, e := range elems {
+		res = append(res, e...)
+	}
+	return res
+}
+
+func randUint64() (r uint64) {
+	buf := make([]byte, 8)
+	_, err := rand.Read(buf)
+	checkFatal(err)
+	for _, v := range buf {
+		r <<= 8
+		r |= uint64(v)
+	}
+	return
+}
+
+func macint(mac net.HardwareAddr) (r uint64) {
+	for _, b := range mac {
+		r <<= 8
+		r |= uint64(b)
+	}
+	return
+}
+
+func intmac(key uint64) (r net.HardwareAddr) {
+	r = make([]byte, 6)
+	for i := 5; i >= 0; i-- {
+		r[i] = byte(key)
+		key >>= 8
+	}
+	return
+}
+
+type ListOfPeers []*Peer
+
+func (lop ListOfPeers) Len() int {
+	return len(lop)
+}
+func (lop ListOfPeers) Swap(i, j int) {
+	lop[i], lop[j] = lop[j], lop[i]
+}
+func (lop ListOfPeers) Less(i, j int) bool {
+	return lop[i].Name < lop[i].Name
+}
+
+// given an address like '1.2.3.4:567', return the address if it has a port,
+// otherwise return the address with weave's standard port number
+func NormalisePeerAddr(peerAddr string) string {
+	_, _, err := net.SplitHostPort(peerAddr)
+	if err == nil {
+		return peerAddr
+	} else {
+		return fmt.Sprintf("%s:%d", peerAddr, Port)
+	}
+}