reweave/weave
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Introduce IntraHost

Browse Source
This commit is contained in:
David Wragg committed 2015-06-12 16:12:54 +01:00
1 parent b77abf480b
commit ef0153cb0f
5 files changed
+125 -94
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

router/intrahost.go
+31
View File
@@ -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 "<no intra-host networking>"
}
router/json.go
+1 -1
View File
@@ -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
}
router/pcap.go
+60 -29
View File
@@ -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)
}
}
router/router.go
+22 -59
View File
@@ -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 {
weaver/main.go
+11 -5
View File
@@ -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)