diff --git a/router/connection.go b/router/connection.go index e2c77bd1..f4cc3762 100644 --- a/router/connection.go +++ b/router/connection.go @@ -167,7 +167,7 @@ func (conn *LocalConnection) log(args ...interface{}) { // ACTOR client API const ( - CSendTCP = iota + CSendProtocolMsg = iota CSetEstablished CReceivedHeartbeat CShutdown @@ -200,10 +200,10 @@ func (conn *LocalConnection) SetEstablished() { } // Async -func (conn *LocalConnection) SendTCP(msg []byte) { +func (conn *LocalConnection) SendProtocolMsg(m ProtocolMsg) { conn.queryChan <- &ConnectionInteraction{ - Interaction: Interaction{code: CSendTCP}, - payload: msg} + Interaction: Interaction{code: CSendProtocolMsg}, + payload: m} } // ACTOR server @@ -265,16 +265,16 @@ func (conn *LocalConnection) queryLoop(queryChan <-chan *ConnectionInteraction) err = conn.handleReceivedHeartbeat(query.payload.(*net.UDPAddr)) case CSetEstablished: err = conn.handleSetEstablished() - case CSendTCP: - err = conn.handleSendTCP(query.payload.([]byte)) + case CSendProtocolMsg: + err = conn.handleSendProtocolMsg(query.payload.(ProtocolMsg)) } case <-tickerChan(conn.heartbeat): conn.Forward(true, conn.heartbeatFrame, nil) case <-tickerChan(conn.fetchAll): - err = conn.handleSendTCP(ProtocolFetchAllByte) + err = conn.handleSendSimpleProtocolMsg(ProtocolFetchAll) case <-tickerChan(conn.fragTest): conn.setStackFrag(false) - err = conn.handleSendTCP(ProtocolStartFragmentationTestByte) + err = conn.handleSendSimpleProtocolMsg(ProtocolStartFragmentationTest) } } return @@ -295,7 +295,7 @@ func (conn *LocalConnection) handleReceivedHeartbeat(remoteUDPAddr *net.UDPAddr) conn.receivedHeartbeat = true conn.Unlock() if !old { - if err := conn.handleSendTCP(ProtocolConnectionEstablishedByte); err != nil { + if err := conn.handleSendSimpleProtocolMsg(ProtocolConnectionEstablished); err != nil { return err } } @@ -332,18 +332,22 @@ func (conn *LocalConnection) handleSetEstablished() error { conn.fragTest = time.NewTicker(FragTestInterval) // avoid initial waits for timers to fire conn.Forward(true, conn.heartbeatFrame, nil) - if err := conn.handleSendTCP(ProtocolFetchAllByte); err != nil { + if err := conn.handleSendSimpleProtocolMsg(ProtocolFetchAll); err != nil { return err } conn.setStackFrag(false) - if err := conn.handleSendTCP(ProtocolStartFragmentationTestByte); err != nil { + if err := conn.handleSendSimpleProtocolMsg(ProtocolStartFragmentationTest); err != nil { return err } return nil } -func (conn *LocalConnection) handleSendTCP(msg []byte) error { - return conn.tcpSender.Send(msg) +func (conn *LocalConnection) handleSendSimpleProtocolMsg(tag ProtocolTag) error { + return conn.handleSendProtocolMsg(ProtocolMsg{tag: tag}) +} + +func (conn *LocalConnection) handleSendProtocolMsg(m ProtocolMsg) error { + return conn.tcpSender.Send(Concat([]byte{byte(m.tag)}, m.msg)) } func (conn *LocalConnection) handleShutdown() { @@ -530,7 +534,7 @@ func (conn *LocalConnection) receiveTCP(decoder *gob.Decoder) { continue } payload := msg[1:] - switch msg[0] { + switch ProtocolTag(msg[0]) { case ProtocolConnectionEstablished: // We sent fast heartbeats to the remote peer, which has // now received at least one of them and told us via this @@ -564,13 +568,13 @@ func (conn *LocalConnection) receiveTCP(decoder *gob.Decoder) { // 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. - conn.SendTCP(Concat(ProtocolUpdateByte, conn.Router.Peers.EncodeAllPeers())) + conn.SendProtocolMsg(ProtocolMsg{ProtocolUpdate, conn.Router.Peers.EncodeAllPeers()}) case ProtocolUpdate: newUpdate, err := conn.Router.Peers.ApplyUpdate(payload) if _, ok := err.(UnknownPeersError); err != nil && ok { // That update contained a peer we didn't know about; // request full update - conn.SendTCP(ProtocolFetchAllByte) + conn.SendProtocolMsg(ProtocolMsg{ProtocolFetchAll, nil}) continue } if conn.CheckFatal(err) != nil { @@ -579,7 +583,7 @@ func (conn *LocalConnection) receiveTCP(decoder *gob.Decoder) { if len(newUpdate) != 0 { conn.Router.ConnectionMaker.Refresh() conn.Router.Routes.Recalculate() - conn.Router.Ourself.BroadcastTCP(Concat(ProtocolUpdateByte, newUpdate)) + conn.Router.Ourself.SendProtocolMsg(ProtocolMsg{ProtocolUpdate, newUpdate}) } case ProtocolPMTUVerified: conn.verifyPMTU <- int(binary.BigEndian.Uint16(payload)) diff --git a/router/consts.go b/router/consts.go index 84baddd0..b15e24a4 100644 --- a/router/consts.go +++ b/router/consts.go @@ -6,8 +6,6 @@ import ( ) const ( - Protocol = "weave" - ProtocolVersion = 10 EthernetOverhead = 14 UDPOverhead = 28 // 20 bytes for IPv4, 8 bytes for UDP Port = 6783 @@ -28,24 +26,7 @@ const ( MaxDuration = time.Duration(math.MaxInt64) ) -const ( - ProtocolConnectionEstablished = iota - ProtocolFragmentationReceived - ProtocolStartFragmentationTest - ProtocolNonce - ProtocolFetchAll - ProtocolUpdate - ProtocolPMTUVerified -) - 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} + FragTest = make([]byte, FragTestSize) + PMTUDiscovery = make([]byte, PMTUDiscoverySize) ) diff --git a/router/crypto.go b/router/crypto.go index ef9c1b4b..8b57722c 100644 --- a/router/crypto.go +++ b/router/crypto.go @@ -85,7 +85,8 @@ func EncodeNonce(df bool) (*[24]byte, []byte, error) { flags = flags | 1 } SetNonceLow15Bits(&nonce, flags) - return &nonce, Concat(ProtocolNonceByte, nonce[:]), nil + // NB: need to make a copy since callers may modify the array + return &nonce, Concat(nonce[:]), nil } func DecodeNonce(msg []byte) (bool, *[24]byte) { @@ -206,7 +207,7 @@ func (ne *NaClEncryptor) Bytes() []byte { if err = ne.conn.CheckFatal(err); err != nil { return []byte{} } - ne.conn.SendTCP(encodedNonce) + ne.conn.SendProtocolMsg(ProtocolMsg{ProtocolNonce, encodedNonce}) ne.nonce = freshNonce nonce = freshNonce } @@ -225,7 +226,7 @@ func (ne *NaClEncryptor) Bytes() []byte { return []byte{} } ne.nonceChan <- nonce - ne.conn.SendTCP(encodedNonce) + ne.conn.SendProtocolMsg(ProtocolMsg{ProtocolNonce, encodedNonce}) } ne.offset = offset diff --git a/router/local_peer.go b/router/local_peer.go index 9c85959f..1b1160f5 100644 --- a/router/local_peer.go +++ b/router/local_peer.go @@ -117,9 +117,9 @@ func (peer *LocalPeer) CreateConnection(peerAddr string, acceptNewPeer bool) err const ( PAddConnection = iota - PBroadcastTCP PDeleteConnection PConnectionEstablished + PSendProtocolMsg ) // Async: rely on the peer to shut us down if we shouldn't be adding @@ -147,10 +147,10 @@ func (peer *LocalPeer) ConnectionEstablished(conn *LocalConnection) { } // Async. -func (peer *LocalPeer) BroadcastTCP(msg []byte) { +func (peer *LocalPeer) SendProtocolMsg(m ProtocolMsg) { peer.queryChan <- &PeerInteraction{ - Interaction: Interaction{code: PBroadcastTCP}, - payload: msg} + Interaction: Interaction{code: PSendProtocolMsg}, + payload: m} } // ACTOR server @@ -169,8 +169,8 @@ func (peer *LocalPeer) queryLoop(queryChan <-chan *PeerInteraction) { query.resultChan <- nil case PConnectionEstablished: peer.handleConnectionEstablished(query.payload.(*LocalConnection)) - case PBroadcastTCP: - peer.handleBroadcastTCP(query.payload.([]byte)) + case PSendProtocolMsg: + peer.handleSendProtocolMsg(query.payload.(ProtocolMsg)) } } } @@ -248,9 +248,9 @@ func (peer *LocalPeer) handleConnectionEstablished(conn *LocalConnection) { peer.broadcastPeerUpdate(conn.Remote()) } -func (peer *LocalPeer) handleBroadcastTCP(msg []byte) { +func (peer *LocalPeer) handleSendProtocolMsg(m ProtocolMsg) { peer.ForEachConnection(func(_ PeerName, conn Connection) { - conn.(*LocalConnection).SendTCP(msg) + conn.(ProtocolSender).SendProtocolMsg(m) }) } @@ -278,7 +278,7 @@ func (peer *LocalPeer) connectionEstablished(conn Connection) { func (peer *LocalPeer) broadcastPeerUpdate(peers ...*Peer) { peer.Router.Routes.Recalculate() - peer.handleBroadcastTCP(Concat(ProtocolUpdateByte, EncodePeers(append(peers, peer.Peer)...))) + peer.handleSendProtocolMsg(ProtocolMsg{ProtocolUpdate, EncodePeers(append(peers, peer.Peer)...)}) } func (peer *LocalPeer) checkConnectionLimit() error { diff --git a/router/protocol.go b/router/protocol.go new file mode 100644 index 00000000..2218d813 --- /dev/null +++ b/router/protocol.go @@ -0,0 +1,27 @@ +package router + +const ( + Protocol = "weave" + ProtocolVersion = 10 +) + +type ProtocolTag byte + +const ( + ProtocolConnectionEstablished ProtocolTag = iota + ProtocolFragmentationReceived + ProtocolStartFragmentationTest + ProtocolNonce + ProtocolFetchAll + ProtocolUpdate + ProtocolPMTUVerified +) + +type ProtocolMsg struct { + tag ProtocolTag + msg []byte +} + +type ProtocolSender interface { + SendProtocolMsg(m ProtocolMsg) +} diff --git a/router/router.go b/router/router.go index d9e31687..bc6170b3 100644 --- a/router/router.go +++ b/router/router.go @@ -308,12 +308,12 @@ func (router *Router) handleUDPPacketFunc(dec *EthernetDecoder, po PacketSink) F case frameLen == EthernetOverhead+8: relayConn.ReceivedHeartbeat(sender, binary.BigEndian.Uint64(frame[EthernetOverhead:])) case frameLen == FragTestSize && bytes.Equal(frame, FragTest): - relayConn.SendTCP(ProtocolFragmentationReceivedByte) + relayConn.SendProtocolMsg(ProtocolMsg{ProtocolFragmentationReceived, nil}) case frameLen == PMTUDiscoverySize && bytes.Equal(frame, PMTUDiscovery): default: frameLenBytes := []byte{0, 0} binary.BigEndian.PutUint16(frameLenBytes, uint16(frameLen-EthernetOverhead)) - relayConn.SendTCP(Concat(ProtocolPMTUVerifiedByte, frameLenBytes)) + relayConn.SendProtocolMsg(ProtocolMsg{ProtocolPMTUVerified, frameLenBytes}) } return nil }