From 57e78e2becbef5bf25883ec09c7e24c8a6fd9985 Mon Sep 17 00:00:00 2001 From: David Wragg Date: Mon, 3 Aug 2015 16:06:02 +0100 Subject: [PATCH] Introduce Bridge --- prog/weaver/main.go | 52 ++++++++++-------- router/bridge.go | 36 +++++++++++++ router/pcap.go | 125 ++++++++++++++++++++++++++++++-------------- router/router.go | 81 +++++++--------------------- router/status.go | 12 +---- 5 files changed, 174 insertions(+), 132 deletions(-) create mode 100644 router/bridge.go diff --git a/prog/weaver/main.go b/prog/weaver/main.go index 1ac9c315..5dac6aa8 100644 --- a/prog/weaver/main.go +++ b/prog/weaver/main.go @@ -36,31 +36,31 @@ func main() { runtime.GOMAXPROCS(procs) var ( - config weave.Config - justVersion bool - protocolMinVersion int - ifaceName string - routerName string - nickName string - password string - pktdebug bool - logLevel string - prof string - bufSzMB int - noDiscovery bool - httpAddr string - iprangeCIDR string - ipsubnetCIDR string - peerCount int - apiPath string - peers []string - + config weave.Config + justVersion bool + protocolMinVersion int + ifaceName string + routerName string + nickName string + password string + pktdebug bool + logLevel string + prof string + bufSzMB int + noDiscovery bool + httpAddr string + iprangeCIDR string + ipsubnetCIDR string + peerCount int + apiPath string + peers []string noDNS bool dnsDomain string dnsListenAddress string dnsTTL int dnsClientTimeout time.Duration dnsEffectiveListenAddress string + iface *net.Interface ) mflag.BoolVar(&justVersion, []string{"#version", "-version"}, false, "print version and exit") @@ -116,18 +116,25 @@ func main() { var err error if ifaceName != "" { - config.Iface, err = weavenet.EnsureInterface(ifaceName) + iface, err := weavenet.EnsureInterface(ifaceName) + if err != nil { + Log.Fatal(err) + } + + // bufsz flag is in MB + config.Bridge, err = weave.NewPcap(iface, bufSzMB*1024*1024) if err != nil { Log.Fatal(err) } } if routerName == "" { - if config.Iface == nil { + if iface == nil { Log.Fatal("Either an interface must be specified with --iface or a name with -name") } - routerName = config.Iface.HardwareAddr.String() + routerName = iface.HardwareAddr.String() } + name, err := weave.PeerNameFromUserInput(routerName) if err != nil { Log.Fatal(err) @@ -157,7 +164,6 @@ func main() { defer profile.Start(&p).Stop() } - config.BufSz = bufSzMB * 1024 * 1024 config.LogFrame = logFrameFunc(pktdebug) config.PeerDiscovery = !noDiscovery diff --git a/router/bridge.go b/router/bridge.go new file mode 100644 index 00000000..9e069177 --- /dev/null +++ b/router/bridge.go @@ -0,0 +1,36 @@ +package router + +// Interface to packet handling on the local virtual bridge +type Bridge interface { + // Inject a packet to be delivered locally + InjectPacket([]byte) error + + // Start consuming packets from the bridge + StartConsumingPackets(BridgeConsumer) error + + String() string + 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) + +type NullBridge struct{} + +func (NullBridge) InjectPacket([]byte) error { + return nil +} + +func (NullBridge) StartConsumingPackets(BridgeConsumer) error { + return nil +} + +func (NullBridge) String() string { + return "" +} + +func (NullBridge) Stats() map[string]int { + return nil +} diff --git a/router/pcap.go b/router/pcap.go index 698c5d0e..138cfcd1 100644 --- a/router/pcap.go +++ b/router/pcap.go @@ -2,18 +2,42 @@ package router import ( "fmt" + "net" + "sync" "github.com/google/gopacket/pcap" ) -type PcapIO struct { - handle *pcap.Handle +type Pcap struct { + iface *net.Interface + bufSz int + + // The libpcap handle for writing packets. It's possible that a + // single handle could be used for reading and writing, but + // we'd have to examine the performance implications. + writeHandle *pcap.Handle + + // pcap handles are single-threaded, so we need to lock around + // uses of writeHandle. + mutex sync.Mutex + + // The libpcap handle for reading packets + readHandle *pcap.Handle } -func NewPcapIO(ifName string, bufSz int) (PacketSourceSink, error) { - pio, err := newPcapIO(ifName, true, 65535, bufSz) +func NewPcap(iface *net.Interface, bufSz int) (Bridge, error) { + wh, err := newPcapHandle(iface.Name, false, 0, 0) if err != nil { - return pio, err + return nil, err + } + + return &Pcap{iface: iface, bufSz: bufSz, writeHandle: wh}, nil +} + +func (p *Pcap) StartConsumingPackets(consumer BridgeConsumer) error { + rh, err := newPcapHandle(p.iface.Name, true, 65535, p.bufSz) + if err != nil { + return err } // Under Linux, libpcap implements the SetDirection filtering @@ -21,16 +45,25 @@ func NewPcapIO(ifName string, bufSz int) (PacketSourceSink, error) { // 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 + err = rh.SetBPFFilter("inbound") + if err != nil { + rh.Close() + return err + } + + // readHandle is just for the benefit of Stats. + p.mutex.Lock() + defer p.mutex.Unlock() + if p.readHandle != nil { + panic("already consuming") + } + + p.readHandle = rh + go p.sniff(rh, consumer) + return nil } -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) { +func newPcapHandle(ifName string, promisc bool, snaplen int, bufSz int) (handle *pcap.Handle, err error) { inactive, err := pcap.NewInactiveHandle(ifName) if err != nil { return @@ -61,38 +94,54 @@ func newPcapIO(ifName string, promisc bool, snaplen int, bufSz int) (handle *Pca if err = inactive.SetBufferSize(bufSz); err != nil { return } - active, err := inactive.Activate() + handle, err = inactive.Activate() if err != nil { return } - if err = active.SetDirection(pcap.DirectionIn); err != nil { - return - } - return &PcapIO{handle: active}, nil -} - -func (pio *PcapIO) ReadPacket() (data []byte, err error) { - for { - data, _, err = pio.handle.ZeroCopyReadPacketData() - if err == nil || err != pcap.NextErrorTimeoutExpired { - break - } - } + err = handle.SetDirection(pcap.DirectionIn) return } -func (pio *PcapIO) WritePacket(data []byte) error { - return pio.handle.WritePacketData(data) +func (p *Pcap) String() string { + return fmt.Sprint(p.iface.Name, " (via pcap)") } -func (pio *PcapIO) Stats() map[string]int { - stats, err := pio.handle.Stats() - if err != nil { - return nil +func (p *Pcap) InjectPacket(pkt []byte) error { + p.mutex.Lock() + defer p.mutex.Unlock() + return p.writeHandle.WritePacketData(pkt) +} + +func (p *Pcap) sniff(readHandle *pcap.Handle, consumer BridgeConsumer) { + dec := NewEthernetDecoder() + + for { + pkt, _, err := readHandle.ZeroCopyReadPacketData() + if err == pcap.NextErrorTimeoutExpired { + continue + } + + checkFatal(err) + dec.DecodeLayers(pkt) + consumer(pkt, dec) } - res := make(map[string]int) - res["PacketsReceived"] = stats.PacketsReceived - res["PacketsDropped"] = stats.PacketsDropped - res["PacketsIfDropped"] = stats.PacketsIfDropped - return res +} + +func (p *Pcap) Stats() map[string]int { + p.mutex.Lock() + rh := p.readHandle + p.mutex.Unlock() + + if rh != nil { + stats, err := rh.Stats() + if err == nil { + return map[string]int{ + "PacketsReceived": stats.PacketsReceived, + "PacketsDropped": stats.PacketsDropped, + "PacketsIfDropped": stats.PacketsIfDropped, + } + } + } + + return nil } diff --git a/router/router.go b/router/router.go index 9268f280..8d845fd9 100644 --- a/router/router.go +++ b/router/router.go @@ -29,12 +29,12 @@ type LogFrameFunc func(string, []byte, *EthernetDecoder) type Config struct { Port int ProtocolMinVersion byte - Iface *net.Interface Password []byte ConnLimit int PeerDiscovery bool BufSz int LogFrame LogFrameFunc + Bridge Bridge } type Router struct { @@ -47,27 +47,17 @@ type Router struct { gossipLock sync.RWMutex gossipChannels GossipChannels TopologyGossip Gossip - PacketSource PacketSource UDPListener *net.UDPConn acceptLimiter *TokenBucket } -type PacketSource interface { - ReadPacket() ([]byte, error) - Stats() map[string]int -} - -type PacketSink interface { - WritePacket([]byte) error -} - -type PacketSourceSink interface { - PacketSource - PacketSink -} - func NewRouter(config Config, name PeerName, nickName string) *Router { router := &Router{Config: config, gossipChannels: make(GossipChannels)} + + if router.Bridge == nil { + router.Bridge = NullBridge{} + } + onMacExpiry := func(mac net.HardwareAddr, peer *Peer) { log.Println("Expired MAC", mac, "at", peer) } @@ -91,22 +81,10 @@ func NewRouter(config Config, name PeerName, nickName string) *Router { // packets forwarded over UDP. This is separate from NewRouter so // that gossipers can register before we start forming connections. func (router *Router) Start() { - // we need two pcap handles since they aren't thread-safe - var pio PacketSourceSink - var po PacketSink - var err error - if router.Iface != nil { - pio, err = NewPcapIO(router.Iface.Name, router.BufSz) - checkFatal(err) - po, err = NewPcapO(router.Iface.Name) - checkFatal(err) - } - router.UDPListener = router.listenUDP(router.Port, po) + log.Println("Sniffing traffic on", router.Bridge) + checkFatal(router.Bridge.StartConsumingPackets(router.handleCapturedPacket)) + router.UDPListener = router.listenUDP(router.Port) router.listenTCP(router.Port) - if pio != nil { - router.PacketSource = pio - router.sniff(pio) - } } func (router *Router) Stop() error { @@ -118,26 +96,8 @@ func (router *Router) UsingPassword() bool { return router.Password != nil } -func (router *Router) sniff(pio PacketSourceSink) { - log.Println("Sniffing traffic on", router.Iface) - - dec := NewEthernetDecoder() - mac := router.Iface.HardwareAddr - if router.Macs.Enter(mac, router.Ourself.Peer) { - log.Println("Discovered our MAC", mac) - } - go func() { - for { - pkt, err := pio.ReadPacket() - checkFatal(err) - router.LogFrame("Sniffed", pkt, nil) - router.handleCapturedPacket(pkt, dec, pio) - } - }() -} - -func (router *Router) handleCapturedPacket(frameData []byte, dec *EthernetDecoder, po PacketSink) { - dec.DecodeLayers(frameData) +func (router *Router) handleCapturedPacket(frameData []byte, dec *EthernetDecoder) { + router.LogFrame("Sniffed", frameData, dec) decodedLen := len(dec.decoded) if decodedLen == 0 { return @@ -179,7 +139,7 @@ func (router *Router) handleCapturedPacket(frameData []byte, dec *EthernetDecode err := router.Ourself.Forward(dstPeer, frameCopy, dec) if ftbe, ok := err.(FrameTooBigError); ok { - err = dec.sendICMPFragNeeded(ftbe.EPMTU, po.WritePacket) + err = dec.sendICMPFragNeeded(ftbe.EPMTU, router.Bridge.InjectPacket) } checkWarn(err) } @@ -213,7 +173,7 @@ func (router *Router) acceptTCP(tcpConn *net.TCPConn) { StartLocalConnection(connRemote, tcpConn, nil, router, true) } -func (router *Router) listenUDP(localPort int, po PacketSink) *net.UDPConn { +func (router *Router) listenUDP(localPort int) *net.UDPConn { localAddr, err := net.ResolveUDPAddr("udp4", fmt.Sprint(":", localPort)) checkFatal(err) conn, err := net.ListenUDP("udp4", localAddr) @@ -225,11 +185,11 @@ func (router *Router) listenUDP(localPort int, po PacketSink) *net.UDPConn { // 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) + go router.udpReader(conn) return conn } -func (router *Router) udpReader(conn *net.UDPConn, po PacketSink) { +func (router *Router) udpReader(conn *net.UDPConn) { defer conn.Close() dec := NewEthernetDecoder() buf := make([]byte, MaxUDPPacketSize) @@ -255,7 +215,7 @@ func (router *Router) udpReader(conn *net.UDPConn, po PacketSink) { if !ok { continue } - if err := relayConn.Decryptor.IterateFrames(packet, router.handleUDPPacketFunc(relayConn, dec, sender, po)); err != nil { + if err := relayConn.Decryptor.IterateFrames(packet, router.handleUDPPacketFunc(relayConn, dec, sender)); err != nil { // Errors during UDP packet decoding / processing are // non-fatal. One common cause is that we receive and // attempt to decrypt a "stray" packet. This can actually @@ -272,7 +232,7 @@ func (router *Router) udpReader(conn *net.UDPConn, po PacketSink) { } } -func (router *Router) handleUDPPacketFunc(relayConn *LocalConnection, dec *EthernetDecoder, sender *net.UDPAddr, po PacketSink) FrameConsumer { +func (router *Router) handleUDPPacketFunc(relayConn *LocalConnection, dec *EthernetDecoder, sender *net.UDPAddr) FrameConsumer { return func(srcNameByte, dstNameByte []byte, frame []byte) { srcPeer := router.Peers.Fetch(PeerNameFromBin(srcNameByte)) dstPeer := router.Peers.Fetch(PeerNameFromBin(dstNameByte)) @@ -320,10 +280,9 @@ func (router *Router) handleUDPPacketFunc(relayConn *LocalConnection, dec *Ether if router.Macs.Enter(srcMac, srcPeer) { log.Println("Discovered remote MAC", srcMac, "at", srcPeer) } - if po != nil { - router.LogFrame("Injecting", frame, dec) - checkWarn(po.WritePacket(frame)) - } + + router.LogFrame("Injecting", frame, dec) + checkWarn(router.Bridge.InjectPacket(frame)) dstPeer, found := router.Macs.Lookup(dstMac) if !found || dstPeer != router.Ourself.Peer { diff --git a/router/status.go b/router/status.go index 833c24d1..7ff69f21 100644 --- a/router/status.go +++ b/router/status.go @@ -63,14 +63,6 @@ type LocalConnectionStatus struct { } func NewStatus(router *Router) *Status { - var ifaceName string - if router.Iface != nil { - ifaceName = router.Iface.Name - } - var captureStats map[string]int - if router.PacketSource != nil { - captureStats = router.PacketSource.Stats() - } return &Status{ Protocol, ProtocolMinVersion, @@ -80,8 +72,8 @@ func NewStatus(router *Router) *Status { router.Ourself.Name.String(), router.Ourself.NickName, router.Port, - ifaceName, - captureStats, + router.Bridge.String(), + router.Bridge.Stats(), NewMACStatusSlice(router.Macs), NewPeerStatusSlice(router.Peers), NewUnicastRouteStatusSlice(router.Routes),