diff --git a/router/intrahost.go b/router/intrahost.go new file mode 100644 index 00000000..dcf15ee9 --- /dev/null +++ b/router/intrahost.go @@ -0,0 +1,31 @@ +package router + +// Interface to intra-host (i.e. container) packet handling +type IntraHost interface { + // Inject a packet to be delivered locally + InjectPacket([]byte) error + + // Start consuming packets + ConsumePackets(IntraHostConsumer) error +} + +type IntraHostConsumer interface { + // A locally captured packet. The caller must supply an + // EthernetDecoder specific to this thread, but it should not + // have been used to decode the packet data. + CapturedPacket([]byte, *EthernetDecoder) +} + +type NullIntraHost struct{} + +func (NullIntraHost) InjectPacket([]byte) error { + return nil +} + +func (NullIntraHost) ConsumePackets(IntraHostConsumer) error { + return nil +} + +func (NullIntraHost) String() string { + return "" +} diff --git a/router/json.go b/router/json.go index af48a012..aa93c63b 100644 --- a/router/json.go +++ b/router/json.go @@ -16,7 +16,7 @@ func (router *Router) StatusJSON(version, encryption string) ([]byte, error) { Macs *MacCache Peers *Peers Routes *Routes - }{version, encryption, router.Ourself.Name.String(), router.Ourself.NickName, fmt.Sprintf("%v", router.Iface), router.Macs, router.Peers, router.Routes}) + }{version, encryption, router.Ourself.Name.String(), router.Ourself.NickName, fmt.Sprint(router.IntraHost), router.Macs, router.Peers, router.Routes}) // leaving out ConectionMaker due to async complexities } diff --git a/router/pcap.go b/router/pcap.go index b7365f68..2cd7570d 100644 --- a/router/pcap.go +++ b/router/pcap.go @@ -3,16 +3,38 @@ package router import ( "code.google.com/p/gopacket/pcap" "fmt" + "log" + "net" + "sync" ) -type PcapIO struct { - handle *pcap.Handle +type Pcap struct { + iface *net.Interface + bufSz int + + // The libpcap handle for writing packets. It's possible thata + // 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. + writeHandleMutex sync.Mutex } -func NewPcapIO(ifName string, bufSz int) (PacketSourceSink, error) { - pio, err := newPcapIO(ifName, true, 65535, bufSz) +func NewPcap(iface *net.Interface, bufSz int) (IntraHost, 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) ConsumePackets(consumer IntraHostConsumer) error { + rh, err := newPcapHandle(p.iface.Name, true, 65535, p.bufSz) + if err != nil { + return err } // Under Linux, libpcap implements the SetDirection filtering @@ -20,16 +42,16 @@ 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 { + return err + } + + 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 @@ -60,26 +82,35 @@ 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 (pi *PcapIO) ReadPacket() (data []byte, err error) { - for { - data, _, err = pi.handle.ZeroCopyReadPacketData() - if err == nil || err != pcap.NextErrorTimeoutExpired { - break - } - } + err = handle.SetDirection(pcap.DirectionIn) return } -func (po *PcapIO) WritePacket(data []byte) error { - return po.handle.WritePacketData(data) +func (p *Pcap) String() string { + return fmt.Sprint(p.iface, "(via pcap)") +} + +func (p *Pcap) InjectPacket(pkt []byte) error { + p.writeHandleMutex.Lock() + defer p.writeHandleMutex.Unlock() + return p.writeHandle.WritePacketData(pkt) +} + +func (p *Pcap) sniff(readHandle *pcap.Handle, consumer IntraHostConsumer) { + log.Println("Sniffing traffic on", p.iface) + dec := NewEthernetDecoder() + + for { + pkt, _, err := readHandle.ZeroCopyReadPacketData() + if err == pcap.NextErrorTimeoutExpired { + continue + } + + checkFatal(err) + consumer.CapturedPacket(pkt, dec) + } } diff --git a/router/router.go b/router/router.go index 11cc3038..eb1790a6 100644 --- a/router/router.go +++ b/router/router.go @@ -20,11 +20,10 @@ type LogFrameFunc func(string, []byte, *EthernetDecoder) type RouterConfig struct { Port int - Iface *net.Interface Password []byte ConnLimit int - BufSz int LogFrame LogFrameFunc + IntraHost IntraHost } type Router struct { @@ -39,21 +38,15 @@ type Router struct { UDPListener *net.UDPConn } -type PacketSource interface { - ReadPacket() ([]byte, error) -} - -type PacketSink interface { - WritePacket([]byte) error -} - -type PacketSourceSink interface { - PacketSource - PacketSink -} - func NewRouter(config RouterConfig, name PeerName, nickName string) *Router { router := &Router{RouterConfig: config, GossipChannels: make(map[uint32]*GossipChannel)} + + // If the caller didn't set the intrahost, replace it with a + // null implementation + if router.IntraHost == nil { + router.IntraHost = NullIntraHost{} + } + onMacExpiry := func(mac net.HardwareAddr, peer *Peer) { log.Println("Expired MAC", mac, "at", peer) } @@ -74,21 +67,9 @@ func NewRouter(config RouterConfig, name PeerName, nickName string) *Router { // Start listening for packets from containers, TCP connections and // packets forwarded over UDP. 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) + checkFatal(router.IntraHost.ConsumePackets(router)) + router.UDPListener = router.listenUDP(router.Port) router.listenTCP(router.Port) - if pio != nil { - router.sniff(pio) - } } func (router *Router) Stop() error { @@ -103,7 +84,7 @@ func (router *Router) UsingPassword() bool { func (router *Router) Status() string { var buf bytes.Buffer fmt.Fprintln(&buf, "Our name is", router.Ourself) - fmt.Fprintln(&buf, "Sniffing traffic on", router.Iface) + fmt.Fprintln(&buf, "Sniffing traffic on", router.IntraHost) fmt.Fprintf(&buf, "MACs:\n%s", router.Macs) fmt.Fprintf(&buf, "Peers:\n%s", router.Peers) fmt.Fprintf(&buf, "Routes:\n%s", router.Routes) @@ -111,25 +92,8 @@ func (router *Router) Status() string { return buf.String() } -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) { +func (router *Router) CapturedPacket(frameData []byte, dec *EthernetDecoder) { + router.LogFrame("Sniffed", frameData, nil) dec.DecodeLayers(frameData) decodedLen := len(dec.decoded) if decodedLen == 0 { @@ -173,7 +137,7 @@ func (router *Router) handleCapturedPacket(frameData []byte, dec *EthernetDecode err := router.Ourself.Forward(dstPeer, df, frameCopy, dec) if ftbe, ok := err.(FrameTooBigError); ok { - err = dec.sendICMPFragNeeded(ftbe.EPMTU, po.WritePacket) + err = dec.sendICMPFragNeeded(ftbe.EPMTU, router.IntraHost.InjectPacket) } checkWarn(err) } @@ -206,7 +170,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) @@ -218,11 +182,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) @@ -248,7 +212,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 @@ -265,7 +229,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, found := router.Peers.Fetch(PeerNameFromBin(srcNameByte)) if !found { @@ -318,10 +282,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.IntraHost.InjectPacket(frame)) dstPeer, found = router.Macs.Lookup(dstMac) if !found || dstPeer != router.Ourself.Peer { diff --git a/weaver/main.go b/weaver/main.go index 7f1bb896..ea0239b9 100644 --- a/weaver/main.go +++ b/weaver/main.go @@ -22,7 +22,6 @@ import ( var version = "(unreleased version)" func main() { - log.SetPrefix(weave.Protocol + " ") log.SetFlags(log.Ldate | log.Ltime | log.Lmicroseconds) @@ -51,6 +50,7 @@ func main() { iprangeCIDR string peerCount int apiPath string + iface *net.Interface ) flag.BoolVar(&justVersion, "version", false, "print version and exit") @@ -84,18 +84,25 @@ func main() { var err error if ifaceName != "" { - config.Iface, err = weavenet.EnsureInterface(ifaceName, wait) + iface, err := weavenet.EnsureInterface(ifaceName, wait) + if err != nil { + log.Fatal(err) + } + + // bufsz flag is in MB + config.IntraHost, 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) @@ -126,7 +133,6 @@ func main() { defer profile.Start(&p).Stop() } - config.BufSz = bufSzMB * 1024 * 1024 config.LogFrame = logFrameFunc(pktdebug) router := weave.NewRouter(config, name, nickName)