diff --git a/prog/weaver/main.go b/prog/weaver/main.go index e7ec13ff..b8bd3326 100644 --- a/prog/weaver/main.go +++ b/prog/weaver/main.go @@ -1,7 +1,6 @@ package main import ( - "crypto/sha256" "fmt" "net" "net/http" @@ -164,9 +163,14 @@ func main() { defer profile.Start(&p).Stop() } - config.Overlay = weave.NewSleeveOverlay(config.Port) - config.LogFrame = logFrameFunc(pktdebug) config.PeerDiscovery = !noDiscovery + config.Overlay = weave.NewSleeveOverlay(config.Port) + + if pktdebug { + config.PacketLogging = packetLogging{} + } else { + config.PacketLogging = nopPacketLogging{} + } router := weave.NewRouter(config, name, nickName) Log.Println("Our name is", router.Ourself) @@ -255,24 +259,22 @@ func canonicalName(f *mflag.Flag) string { return "" } -func logFrameFunc(debug bool) weave.LogFrameFunc { - if !debug { - return func(prefix string, frame []byte, dec *weave.EthernetDecoder) {} - } - return func(prefix string, frame []byte, dec *weave.EthernetDecoder) { - h := fmt.Sprintf("%x", sha256.Sum256(frame)) - parts := []interface{}{prefix, len(frame), "bytes (", h, ")"} +type packetLogging struct{} - if dec != nil { - parts = append(parts, dec.Eth.SrcMAC, "->", dec.Eth.DstMAC) +func (packetLogging) LogPacket(msg string, key weave.PacketKey) { + Log.Println(msg, key.SrcMAC, "->", key.DstMAC) +} - if dec.DF() { - parts = append(parts, "(DF)") - } - } +func (packetLogging) LogForwardPacket(msg string, key weave.ForwardPacketKey) { + Log.Println(msg, key.SrcPeer, key.SrcMAC, "->", key.DstPeer, key.DstMAC) +} - Log.Println(parts...) - } +type nopPacketLogging struct{} + +func (nopPacketLogging) LogPacket(string, weave.PacketKey) { +} + +func (nopPacketLogging) LogForwardPacket(string, weave.ForwardPacketKey) { } func parseAndCheckCIDR(cidrStr string) address.CIDR { diff --git a/router/bridge.go b/router/bridge.go index 9e069177..603d8687 100644 --- a/router/bridge.go +++ b/router/bridge.go @@ -3,7 +3,7 @@ package router // Interface to packet handling on the local virtual bridge type Bridge interface { // Inject a packet to be delivered locally - InjectPacket([]byte) error + InjectPacket(PacketKey) FlowOp // Start consuming packets from the bridge StartConsumingPackets(BridgeConsumer) error @@ -12,14 +12,12 @@ type Bridge interface { Stats() map[string]int } -// A function that accepts locally captured packets. The ethernet -// decoder is specific to this thread, and will already have been used -// to to decode the packet data. -type BridgeConsumer func([]byte, *EthernetDecoder) +// A function that determines how to handle locally captured packets. +type BridgeConsumer func(PacketKey) FlowOp type NullBridge struct{} -func (NullBridge) InjectPacket([]byte) error { +func (NullBridge) InjectPacket(PacketKey) FlowOp { return nil } diff --git a/router/connection.go b/router/connection.go index 8a844cb7..e3967ec6 100644 --- a/router/connection.go +++ b/router/connection.go @@ -446,8 +446,8 @@ func (conn *LocalConnection) extendReadDeadline() { conn.TCPConn.SetReadDeadline(time.Now().Add(TCPHeartbeat * 2)) } -func (conn *LocalConnection) Forward(srcPeer, dstPeer *Peer, frame []byte, dec *EthernetDecoder, broadcast bool) { - conn.forwarder.Forward(srcPeer, dstPeer, frame, dec, broadcast) +func (conn *LocalConnection) Forward(key ForwardPacketKey) FlowOp { + return conn.forwarder.Forward(key) } func tickerChan(ticker *time.Ticker) <-chan time.Time { diff --git a/router/ethernet_decoder.go b/router/ethernet_decoder.go index ebc9b252..2eb1a429 100644 --- a/router/ethernet_decoder.go +++ b/router/ethernet_decoder.go @@ -29,6 +29,12 @@ func (dec *EthernetDecoder) DecodeLayers(data []byte) { dec.parser.DecodeLayers(data, &dec.decoded) } +func (dec *EthernetDecoder) PacketKey() (key PacketKey) { + copy(key.SrcMAC[:], dec.Eth.SrcMAC) + copy(key.DstMAC[:], dec.Eth.DstMAC) + return +} + func (dec *EthernetDecoder) makeICMPFragNeeded(mtu int) ([]byte, error) { buf := gopacket.NewSerializeBuffer() opts := gopacket.SerializeOptions{ @@ -65,15 +71,9 @@ func (dec *EthernetDecoder) makeICMPFragNeeded(mtu int) ([]byte, error) { } var ( - // see http://en.wikipedia.org/wiki/Multicast_address#Ethernet - stpMACPrefix = []byte{0x01, 0x80, 0xC2, 0x00, 0x00} - zeroMAC, _ = net.ParseMAC("00:00:00:00:00:00") + zeroMAC, _ = net.ParseMAC("00:00:00:00:00:00") ) -func (dec *EthernetDecoder) DropFrame() bool { - return bytes.Equal(stpMACPrefix, dec.Eth.DstMAC[:len(stpMACPrefix)]) -} - func (dec *EthernetDecoder) IsSpecial() bool { return dec.Eth.Length == 0 && dec.Eth.EthernetType == layers.EthernetTypeLLC && bytes.Equal(zeroMAC, dec.Eth.SrcMAC) && bytes.Equal(zeroMAC, dec.Eth.DstMAC) diff --git a/router/flow.go b/router/flow.go new file mode 100644 index 00000000..2a7ac039 --- /dev/null +++ b/router/flow.go @@ -0,0 +1,50 @@ +package router + +import "net" + +// Just enough flow machinery for the weave router + +type MAC [6]byte + +func (mac MAC) String() string { + return net.HardwareAddr(mac[:]).String() +} + +type PacketKey struct { + SrcMAC MAC + DstMAC MAC +} + +type ForwardPacketKey struct { + SrcPeer *Peer + DstPeer *Peer + PacketKey +} + +type FlowOp interface { + // The caller must supply an EthernetDecoder specific to this + // thread, which has already been used to decode the frame. + // The broadcast parameter is a hint whether the packet is + // being broadcast. + Process(frame []byte, dec *EthernetDecoder, broadcast bool) +} + +type MultiFlowOp struct { + broadcast bool + ops []FlowOp +} + +func NewMultiFlowOp(broadcast bool) *MultiFlowOp { + return &MultiFlowOp{broadcast: broadcast} +} + +func (mfop *MultiFlowOp) Add(ops ...FlowOp) { + mfop.ops = append(mfop.ops, ops...) +} + +func (mfop *MultiFlowOp) Process(frame []byte, dec *EthernetDecoder, + broadcast bool) { + for _, op := range mfop.ops { + op.Process(frame, dec, mfop.broadcast) + } +} diff --git a/router/local_peer.go b/router/local_peer.go index 7f331556..3a57e8a4 100644 --- a/router/local_peer.go +++ b/router/local_peer.go @@ -27,40 +27,54 @@ func NewLocalPeer(name PeerName, nickName string, router *Router) *LocalPeer { return peer } -func (peer *LocalPeer) Forward(dstPeer *Peer, frame []byte, dec *EthernetDecoder) { - peer.Relay(peer.Peer, dstPeer, frame, dec) +func (peer *LocalPeer) Forward(dstPeer *Peer, key PacketKey) FlowOp { + return peer.Relay(ForwardPacketKey{ + PacketKey: key, + SrcPeer: peer.Peer, + DstPeer: dstPeer, + }) } -func (peer *LocalPeer) Broadcast(frame []byte, dec *EthernetDecoder) { - peer.RelayBroadcast(peer.Peer, frame, dec) +func (peer *LocalPeer) Broadcast(key PacketKey) FlowOp { + return peer.RelayBroadcast(peer.Peer, key) } -func (peer *LocalPeer) Relay(srcPeer, dstPeer *Peer, frame []byte, dec *EthernetDecoder) { - relayPeerName, found := peer.router.Routes.Unicast(dstPeer.Name) +func (peer *LocalPeer) Relay(key ForwardPacketKey) FlowOp { + relayPeerName, found := peer.router.Routes.Unicast(key.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) - return + log.Println("Received packet for unknown destination:", key.DstPeer) + 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 + return nil } - conn.(*LocalConnection).Forward(srcPeer, dstPeer, frame, dec, false) + + return conn.(*LocalConnection).Forward(key) } -func (peer *LocalPeer) RelayBroadcast(srcPeer *Peer, frame []byte, dec *EthernetDecoder) { +func (peer *LocalPeer) RelayBroadcast(srcPeer *Peer, key PacketKey) FlowOp { nextHops := peer.router.Routes.Broadcast(srcPeer.Name) if len(nextHops) == 0 { - return + return nil } + + op := NewMultiFlowOp(true) + for _, conn := range peer.ConnectionsTo(nextHops) { - conn.(*LocalConnection).Forward(srcPeer, conn.Remote(), frame, - dec, true) + op.Add(conn.(*LocalConnection).Forward(ForwardPacketKey{ + PacketKey: key, + SrcPeer: srcPeer, + DstPeer: conn.Remote(), + })) } + + return op } func (peer *LocalPeer) Connections() ConnectionSet { diff --git a/router/mac_cache.go b/router/mac_cache.go index 021030c9..a798c48a 100644 --- a/router/mac_cache.go +++ b/router/mac_cache.go @@ -56,15 +56,15 @@ func (cache *MacCache) Enter(mac net.HardwareAddr, peer *Peer) bool { return false } -func (cache *MacCache) Lookup(mac net.HardwareAddr) (*Peer, bool) { +func (cache *MacCache) Lookup(mac net.HardwareAddr) *Peer { key := macint(mac) cache.RLock() defer cache.RUnlock() entry, found := cache.table[key] if !found { - return nil, false + return nil } - return entry.peer, true + return entry.peer } func (cache *MacCache) Delete(peer *Peer) bool { diff --git a/router/overlay.go b/router/overlay.go index b8ac0699..c8c8d0c7 100644 --- a/router/overlay.go +++ b/router/overlay.go @@ -39,8 +39,7 @@ type ForwarderParams struct { // When a consumer is called, the decoder will already have been used // to decode the frame. -type OverlayConsumer func(src *Peer, dst *Peer, frame []byte, - dec *EthernetDecoder) +type OverlayConsumer func(ForwardPacketKey) FlowOp // Crypto settings for a forwarder. type OverlayCrypto struct { @@ -58,11 +57,8 @@ type OverlayForwarder interface { // (e.g. on another thread). SetListener(OverlayForwarderListener) - // Forward a packet across the connection. The caller must - // supply an EthernetDecoder specific to this thread, which - // has already been used to decode the frame. - Forward(src *Peer, dest *Peer, frame []byte, dec *EthernetDecoder, - broadcast bool) + // Forward a packet across the connection. + Forward(ForwardPacketKey) FlowOp Stop() @@ -88,8 +84,8 @@ func (NullOverlay) MakeForwarder(ForwarderParams) (OverlayForwarder, error) { func (NullOverlay) SetListener(OverlayForwarderListener) { } -func (NullOverlay) Forward(*Peer, *Peer, []byte, *EthernetDecoder, - bool) { +func (NullOverlay) Forward(ForwardPacketKey) FlowOp { + return nil } func (NullOverlay) Stop() { diff --git a/router/pcap.go b/router/pcap.go index 138cfcd1..cddc366e 100644 --- a/router/pcap.go +++ b/router/pcap.go @@ -106,10 +106,14 @@ func (p *Pcap) String() string { return fmt.Sprint(p.iface.Name, " (via pcap)") } -func (p *Pcap) InjectPacket(pkt []byte) error { +func (p *Pcap) InjectPacket(PacketKey) FlowOp { + return p +} + +func (p *Pcap) Process(frame []byte, dec *EthernetDecoder, broadcast bool) { p.mutex.Lock() defer p.mutex.Unlock() - return p.writeHandle.WritePacketData(pkt) + checkWarn(p.writeHandle.WritePacketData(frame)) } func (p *Pcap) sniff(readHandle *pcap.Handle, consumer BridgeConsumer) { @@ -123,7 +127,21 @@ func (p *Pcap) sniff(readHandle *pcap.Handle, consumer BridgeConsumer) { checkFatal(err) dec.DecodeLayers(pkt) - consumer(pkt, dec) + if len(dec.decoded) == 0 { + continue + } + + if fop := consumer(dec.PacketKey()); fop != nil { + // 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 + pktLen := len(pkt) + pktCopy := make([]byte, pktLen, pktLen) + copy(pktCopy, pkt) + + fop.Process(pktCopy, dec, false) + } } } diff --git a/router/router.go b/router/router.go index a26745f0..85140055 100644 --- a/router/router.go +++ b/router/router.go @@ -20,8 +20,6 @@ const ( // [3] control rate at which new tokens are added to the bucket -type LogFrameFunc func(string, []byte, *EthernetDecoder) - type Config struct { Port int ProtocolMinVersion byte @@ -29,11 +27,16 @@ type Config struct { ConnLimit int PeerDiscovery bool BufSz int - LogFrame LogFrameFunc + PacketLogging PacketLogging Bridge Bridge Overlay Overlay } +type PacketLogging interface { + LogPacket(string, PacketKey) + LogForwardPacket(string, ForwardPacketKey) +} + type Router struct { Config Ourself *LocalPeer @@ -97,48 +100,45 @@ func (router *Router) UsingPassword() bool { return router.Password != nil } -func (router *Router) handleCapturedPacket(frameData []byte, dec *EthernetDecoder) { - router.LogFrame("Sniffed", frameData, dec) - decodedLen := len(dec.decoded) - if decodedLen == 0 { - return - } - srcMac := dec.Eth.SrcMAC - srcPeer, found := router.Macs.Lookup(srcMac) +func (router *Router) handleCapturedPacket(key PacketKey) FlowOp { + router.PacketLogging.LogPacket("Captured", key) + srcMac := net.HardwareAddr(key.SrcMAC[:]) + srcPeer := 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.Peer { - return + if srcPeer != nil && srcPeer != router.Ourself.Peer { + return nil } + if router.Macs.Enter(srcMac, router.Ourself.Peer) { log.Println("Discovered local MAC", srcMac) } - if dec.DropFrame() { - return - } - dstMac := dec.Eth.DstMAC - dstPeer, found := router.Macs.Lookup(dstMac) - if found && dstPeer == router.Ourself.Peer { - return - } - router.LogFrame("Forwarding", frameData, dec) - // 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 we don't know which peer corresponds to the dest MAC, - // broadcast it. - if !found { - router.Ourself.Broadcast(frameCopy, dec) - return + // Discard STP broadcasts + if key.DstMAC == [...]byte{0x01, 0x80, 0xC2, 0x00, 0x00, 0x00} { + return nil } - router.Ourself.Forward(dstPeer, frameCopy, dec) + dstMac := net.HardwareAddr(key.DstMAC[:]) + switch dstPeer := router.Macs.Lookup(dstMac); dstPeer { + case router.Ourself.Peer: + // The packet is destined for a local MAC. The bridge + // won't normally send us such packets, and if it does + // it's likely to be broadcasting the packet to all + // ports. So if it happens, just drop the packet to + // avoid warnings if we try to forward it. + return nil + case nil: + // If we don't know which peer corresponds to the dest + // MAC, broadcast it. + router.PacketLogging.LogPacket("Broadcasting", key) + return router.Ourself.Broadcast(key) + default: + router.PacketLogging.LogPacket("Forwarding", key) + return router.Ourself.Forward(dstPeer, key) + } } func (router *Router) listenTCP(localPort int) { @@ -170,29 +170,44 @@ func (router *Router) acceptTCP(tcpConn *net.TCPConn) { StartLocalConnection(connRemote, tcpConn, nil, router, true) } -func (router *Router) handleForwardedPacket(srcPeer *Peer, dstPeer *Peer, - frame []byte, dec *EthernetDecoder) { - if dstPeer != router.Ourself.Peer { +func (router *Router) handleForwardedPacket(key ForwardPacketKey) FlowOp { + if key.DstPeer != router.Ourself.Peer { // it's not for us, we're just relaying it - router.LogFrame("Relaying", frame, dec) - router.Ourself.Relay(srcPeer, dstPeer, frame, dec) - return + router.PacketLogging.LogForwardPacket("Relaying", key) + return router.Ourself.Relay(key) } - srcMac := dec.Eth.SrcMAC - dstMac := dec.Eth.DstMAC + // At this point, it's either unicast to us, or a broadcast + // (because the DstPeer on a forwarded broadcast packet is + // always set to the peer being forwarded to) - if router.Macs.Enter(srcMac, srcPeer) { - log.Println("Discovered remote MAC", srcMac, "at", srcPeer) + srcMac := net.HardwareAddr(key.SrcMAC[:]) + dstMac := net.HardwareAddr(key.DstMAC[:]) + if router.Macs.Enter(srcMac, key.SrcPeer) { + log.Println("Discovered remote MAC", srcMac, "at", key.SrcPeer) } - router.LogFrame("Injecting", frame, dec) - checkWarn(router.Bridge.InjectPacket(frame)) + router.PacketLogging.LogForwardPacket("Injecting", key) + injectFop := router.Bridge.InjectPacket(key.PacketKey) + dstPeer := router.Macs.Lookup(dstMac) + if dstPeer == router.Ourself.Peer { + return injectFop + } - dstPeer, found := router.Macs.Lookup(dstMac) - if !found || dstPeer != router.Ourself.Peer { - router.LogFrame("Relaying broadcast", frame, dec) - router.Ourself.RelayBroadcast(srcPeer, frame, dec) + router.PacketLogging.LogForwardPacket("Relaying broadcast", key) + relayFop := router.Ourself.RelayBroadcast(key.SrcPeer, key.PacketKey) + switch { + case injectFop == nil: + return relayFop + + case relayFop == nil: + return injectFop + + default: + mfop := NewMultiFlowOp(false) + mfop.Add(injectFop) + mfop.Add(relayFop) + return mfop } } diff --git a/router/sleeve.go b/router/sleeve.go index 0c002bd8..f844f4c1 100644 --- a/router/sleeve.go +++ b/router/sleeve.go @@ -231,11 +231,23 @@ func (sleeve *SleeveOverlay) handleFrame(sender *net.UDPAddr, return } + sleeve.sendToConsumer(srcPeer, dstPeer, frame, dec) +} + +func (sleeve *SleeveOverlay) sendToConsumer(srcPeer, dstPeer *Peer, + frame []byte, dec *EthernetDecoder) { if sleeve.consumer == nil { return } - sleeve.consumer(srcPeer, dstPeer, frame, dec) + fop := sleeve.consumer(ForwardPacketKey{ + SrcPeer: srcPeer, + DstPeer: dstPeer, + PacketKey: dec.PacketKey(), + }) + if fop != nil { + fop.Process(frame, dec, false) + } } type udpSender interface { @@ -389,8 +401,18 @@ func (fwd *sleeveForwarder) SetListener(listener OverlayForwarderListener) { } } -func (fwd *sleeveForwarder) Forward(src *Peer, dst *Peer, frame []byte, - dec *EthernetDecoder, broadcast bool) { +type curriedForward struct { + fwd *sleeveForwarder + key ForwardPacketKey +} + +func (fwd *sleeveForwarder) Forward(key ForwardPacketKey) FlowOp { + return curriedForward{fwd, key} +} + +func (f curriedForward) Process(frame []byte, dec *EthernetDecoder, + broadcast bool) { + fwd := f.fwd fwd.lock.RLock() haveContact := (fwd.remoteAddr != nil) mtu := fwd.mtu @@ -403,8 +425,8 @@ func (fwd *sleeveForwarder) Forward(src *Peer, dst *Peer, frame []byte, return } - srcName := src.NameByte - dstName := dst.NameByte + srcName := f.key.SrcPeer.NameByte + dstName := f.key.DstPeer.NameByte // We could use non-blocking channel sends here, i.e. drop frames // on the floor when the forwarder is busy. This would allow our @@ -444,7 +466,8 @@ func (fwd *sleeveForwarder) Forward(src *Peer, dst *Peer, frame []byte, // The frag-needed packet does not have DF set, so the // potential recursion here is bounded. - fwd.sleeve.consumer(dst, src, fragNeededPacket, dec) + fwd.sleeve.sendToConsumer(f.key.DstPeer, f.key.SrcPeer, + fragNeededPacket, dec) return }