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

Introduce Bridge

Browse Source
This commit is contained in:
David Wragg committed 2015-09-08 10:44:37 +01:00
1 parent 64e3cf04cc
commit 57e78e2bec
5 files changed
+174 -132
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

prog/weaver/main.go
+29 -23
View File
@@ -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
router/bridge.go
+36
View File
@@ -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 "<no bridge networking>"
}
func (NullBridge) Stats() map[string]int {
return nil
}
router/pcap.go
+87 -38
View File
@@ -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
}
router/router.go
+20 -61
View File
@@ -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 {
router/status.go
+2 -10
View File
@@ -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),