diff --git a/prog/weaver/http.go b/prog/weaver/http.go index a53f59c8..8871468e 100644 --- a/prog/weaver/http.go +++ b/prog/weaver/http.go @@ -146,16 +146,16 @@ var dnsEntriesTemplate = defTemplate("dnsEntries", `\ type WeaveStatus struct { Version string - Router *weave.Status `json:"Router,omitempty"` - IPAM *ipam.Status `json:"IPAM,omitempty"` - DNS *nameserver.Status `json:"DNS,omitempty"` + Router *weave.NetworkRouterStatus `json:"Router,omitempty"` + IPAM *ipam.Status `json:"IPAM,omitempty"` + DNS *nameserver.Status `json:"DNS,omitempty"` } -func HandleHTTP(muxRouter *mux.Router, version string, router *weave.Router, allocator *ipam.Allocator, defaultSubnet address.CIDR, ns *nameserver.Nameserver, dnsserver *nameserver.DNSServer) { +func HandleHTTP(muxRouter *mux.Router, version string, router *weave.NetworkRouter, allocator *ipam.Allocator, defaultSubnet address.CIDR, ns *nameserver.Nameserver, dnsserver *nameserver.DNSServer) { status := func() WeaveStatus { return WeaveStatus{ version, - weave.NewStatus(router), + weave.NewNetworkRouterStatus(router), ipam.NewStatus(allocator, defaultSubnet), nameserver.NewStatus(ns, dnsserver)} } diff --git a/prog/weaver/main.go b/prog/weaver/main.go index 48bb26df..e925cb2e 100644 --- a/prog/weaver/main.go +++ b/prog/weaver/main.go @@ -42,6 +42,7 @@ func main() { addDatapathInterface string config weave.Config + networkConfig weave.NetworkConfig protocolMinVersion int ifaceName string routerName string @@ -150,7 +151,7 @@ func main() { } config.ProtocolMinVersion = byte(protocolMinVersion) - var fastDPOverlay weave.Overlay + var fastDPOverlay weave.NetworkOverlay if datapathName != "" { // A datapath name implies that "Bridge" and "Overlay" // packet handling use fast datapath, although other @@ -163,7 +164,7 @@ func main() { }) checkFatal(err) - config.Bridge = fastdp.Bridge() + networkConfig.Bridge = fastdp.Bridge() fastDPOverlay = fastdp.Overlay() } @@ -178,7 +179,7 @@ func main() { checkFatal(err) // bufsz flag is in MB - config.Bridge, err = weave.NewPcap(iface, bufSzMB*1024*1024) + networkConfig.Bridge, err = weave.NewPcap(iface, bufSzMB*1024*1024) checkFatal(err) } @@ -229,12 +230,12 @@ func main() { config.PeerDiscovery = !noDiscovery if pktdebug { - config.PacketLogging = packetLogging{} + networkConfig.PacketLogging = packetLogging{} } else { - config.PacketLogging = nopPacketLogging{} + networkConfig.PacketLogging = nopPacketLogging{} } - router := weave.NewRouter(config, name, nickName, overlays) + router := weave.NewNetworkRouter(config, networkConfig, name, nickName, overlays) Log.Println("Our name is", router.Ourself) var dockerCli *docker.Client @@ -258,7 +259,7 @@ func main() { var allocator *ipam.Allocator var defaultSubnet address.CIDR if iprangeCIDR != "" { - allocator, defaultSubnet = createAllocator(router, iprangeCIDR, ipsubnetCIDR, determineQuorum(peerCount, peers), isKnownPeer) + allocator, defaultSubnet = createAllocator(router.Router, iprangeCIDR, ipsubnetCIDR, determineQuorum(peerCount, peers), isKnownPeer) observeContainers(allocator) } else if peerCount > 0 { Log.Fatal("--init-peer-count flag specified without --ipalloc-range") diff --git a/router/connection.go b/router/connection.go index 77a539cc..1cef75b9 100644 --- a/router/connection.go +++ b/router/connection.go @@ -49,7 +49,7 @@ type LocalConnection struct { uid uint64 actionChan chan<- ConnectionAction finished <-chan struct{} // closed to signal that actorLoop has finished - forwarder OverlayForwarder + OverlayConn OverlayConnection } type ConnectionAction func() error @@ -190,7 +190,7 @@ func (conn *LocalConnection) run(actionChan <-chan ConnectionAction, finished ch conn.Log("connection ready; using protocol version", conn.version) - params := ForwarderParams{ + params := OverlayConnectionParams{ RemotePeer: conn.remote, LocalAddr: conn.TCPConn.LocalAddr().(*net.TCPAddr), RemoteAddr: conn.TCPConn.RemoteAddr().(*net.TCPAddr), @@ -200,28 +200,27 @@ func (conn *LocalConnection) run(actionChan <-chan ConnectionAction, finished ch SendControlMessage: conn.sendOverlayControlMessage, Features: intro.Features, } - if conn.forwarder, err = conn.Router.Overlay.MakeForwarder(params); err != nil { + if conn.OverlayConn, err = conn.Router.Overlay.PrepareConnection(params); err != nil { return } // As soon as we do AddConnection, the new connection becomes // visible to the packet routing logic. So AddConnection must - // come after MakeForwarder + // come after PrepareConnection if err = conn.Router.Ourself.AddConnection(conn); err != nil { return } conn.Router.ConnectionMaker.ConnectionCreated(conn) - // Forwarder confirmation comes after AddConnection, because - // only after that completes do we know the connection is - // valid: in particular that it is not a duplicate connection - // to the same peer. Sending heartbeats on a duplicate - // connection can trip up crypto at the other end, since the - // associated UDP packets may get decoded by the other - // connection. It is also generally wasteful to engage in any - // interaction with the remote on a connection that turns out - // to be invalid. - conn.forwarder.Confirm() + // OverlayConnection confirmation comes after AddConnection, + // because only after that completes do we know the connection is + // valid: in particular that it is not a duplicate connection to + // the same peer. Overlay communication on a duplicate connection + // can cause problems such as tripping up overlay crypto at the + // other end due to data being decoded by the other connection. It + // is also generally wasteful to engage in any interaction with + // the remote on a connection that turns out to be invalid. + conn.OverlayConn.Confirm() // receiveTCP must follow also AddConnection. In the absence // of any indirect connectivity to the remote peer, the first @@ -337,8 +336,8 @@ func (conn *LocalConnection) registerRemote(remote *Peer, acceptNewPeer bool) er } func (conn *LocalConnection) actorLoop(actionChan <-chan ConnectionAction) (err error) { - fwdErrorChan := conn.forwarder.ErrorChannel() - fwdEstablishedChan := conn.forwarder.EstablishedChannel() + fwdErrorChan := conn.OverlayConn.ErrorChannel() + fwdEstablishedChan := conn.OverlayConn.EstablishedChannel() for err == nil { select { @@ -379,8 +378,8 @@ func (conn *LocalConnection) shutdown(err error) { conn.heartbeatTCP.Stop() } - if conn.forwarder != nil { - conn.forwarder.Stop() + if conn.OverlayConn != nil { + conn.OverlayConn.Stop() } conn.Router.ConnectionMaker.ConnectionTerminated(conn, err) @@ -424,7 +423,7 @@ func (conn *LocalConnection) handleProtocolMsg(tag ProtocolTag, payload []byte) switch tag { case ProtocolHeartbeat: case ProtocolConnectionEstablished, ProtocolFragmentationReceived, ProtocolPMTUVerified, ProtocolOverlayControlMsg: - conn.forwarder.ControlMessage(byte(tag), payload) + conn.OverlayConn.ControlMessage(byte(tag), payload) case ProtocolGossipUnicast, ProtocolGossipBroadcast, ProtocolGossip: return conn.Router.handleGossip(tag, payload) default: diff --git a/router/fastdp.go b/router/fastdp.go index 440a93b8..b1765a94 100644 --- a/router/fastdp.go +++ b/router/fastdp.go @@ -297,7 +297,7 @@ type fastDatapathOverlay struct { *FastDatapath } -func (fastdp *FastDatapath) Overlay() Overlay { +func (fastdp *FastDatapath) Overlay() NetworkOverlay { return fastDatapathOverlay{fastdp} } @@ -506,7 +506,7 @@ type fastDatapathForwarder struct { errorChan chan error } -func (fastdp fastDatapathOverlay) MakeForwarder(params ForwarderParams) (OverlayForwarder, error) { +func (fastdp fastDatapathOverlay) PrepareConnection(params OverlayConnectionParams) (OverlayConnection, error) { if params.SessionKey != nil { // No encryption suport in fastdp. The weaver main.go // is responsible for ensuring this doesn't happen. diff --git a/router/network_overlay.go b/router/network_overlay.go new file mode 100644 index 00000000..8d80594b --- /dev/null +++ b/router/network_overlay.go @@ -0,0 +1,46 @@ +package router + +// Interface to overlay network packet handling +type NetworkOverlay interface { + Overlay + + // The routes have changed, so any cached information should + // be discarded. + InvalidateRoutes() + + // A mapping of a short id to a peer has changed + InvalidateShortIDs() + + // Start consuming forwarded packets. + StartConsumingPackets(*Peer, *Peers, OverlayConsumer) error +} + +// When a consumer is called, the decoder will already have been used +// to decode the frame. +type OverlayConsumer func(ForwardPacketKey) FlowOp + +// All of the machinery to forward packets to a particular peer +type OverlayForwarder interface { + OverlayConnection + // Forward a packet across the connection. May be called as soon + // as the overlay connection is created, in particular before + // Confirm(). The return value nil means the key could not be + // handled by this forwarder. + Forward(ForwardPacketKey) FlowOp +} + +type NullNetworkOverlay struct{ NullOverlay } + +func (NullNetworkOverlay) InvalidateRoutes() { +} + +func (NullNetworkOverlay) InvalidateShortIDs() { +} + +func (NullNetworkOverlay) StartConsumingPackets(*Peer, *Peers, OverlayConsumer) error { + return nil +} + +func (NullNetworkOverlay) Forward(ForwardPacketKey) FlowOp { + return DiscardingFlowOp{} +} diff --git a/router/network_router.go b/router/network_router.go new file mode 100644 index 00000000..02bff759 --- /dev/null +++ b/router/network_router.go @@ -0,0 +1,196 @@ +package router + +import ( + "net" + "time" +) + +const ( + MaxUDPPacketSize = 65535 + FastHeartbeat = 500 * time.Millisecond + SlowHeartbeat = 10 * time.Second + MaxMissedHeartbeats = 6 + HeartbeatTimeout = MaxMissedHeartbeats * SlowHeartbeat + // should be greater than typical ARP cache expiries, i.e. > 3/2 * + // /proc/sys/net/ipv4_neigh/*/base_reachable_time_ms on Linux + macMaxAge = 10 * time.Minute +) + +type NetworkConfig struct { + BufSz int + PacketLogging PacketLogging + Bridge Bridge +} + +type PacketLogging interface { + LogPacket(string, PacketKey) + LogForwardPacket(string, ForwardPacketKey) +} + +type NetworkRouter struct { + *Router + NetworkConfig + Macs *MacCache +} + +func NewNetworkRouter(config Config, networkConfig NetworkConfig, name PeerName, nickName string, overlay NetworkOverlay) *NetworkRouter { + if overlay == nil { + overlay = NullNetworkOverlay{} + } + if networkConfig.Bridge == nil { + networkConfig.Bridge = NullBridge{} + } + + router := &NetworkRouter{Router: NewRouter(config, name, nickName, overlay), NetworkConfig: networkConfig} + router.Peers.OnInvalidateShortIDs(overlay.InvalidateShortIDs) + router.Routes.OnChange(overlay.InvalidateRoutes) + router.Macs = NewMacCache(macMaxAge, + func(mac net.HardwareAddr, peer *Peer) { + log.Println("Expired MAC", mac, "at", peer) + }) + router.Peers.OnGC(func(peer *Peer) { router.Macs.Delete(peer) }) + return router +} + +// Start listening for TCP connections, locally captured packets, and +// forwarded packets. +func (router *NetworkRouter) Start() { + log.Println("Sniffing traffic on", router.Bridge) + checkFatal(router.Bridge.StartConsumingPackets(router.handleCapturedPacket)) + checkFatal(router.Overlay.(NetworkOverlay).StartConsumingPackets(router.Ourself.Peer, router.Peers, router.handleForwardedPacket)) + router.Router.Start() +} + +func (router *NetworkRouter) handleCapturedPacket(key PacketKey) FlowOp { + router.PacketLogging.LogPacket("Captured", key) + srcMac := net.HardwareAddr(key.SrcMAC[:]) + + switch newSrcMac, conflictPeer := router.Macs.Add(srcMac, router.Ourself.Peer); { + case newSrcMac: + log.Println("Discovered local MAC", srcMac) + + case conflictPeer != nil: + // The MAC cache has an entry for the source MAC + // associated with another peer. This probably means + // we are seeing a frame we injected ourself. That + // shouldn't happen, but discard it just in case. + log.Error("Captured frame from MAC (", srcMac, ") associated with another peer ", conflictPeer) + return DiscardingFlowOp{} + } + + // Discard STP broadcasts + if key.DstMAC == [...]byte{0x01, 0x80, 0xC2, 0x00, 0x00, 0x00} { + return DiscardingFlowOp{} + } + + 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 DiscardingFlowOp{} + case nil: + // If we don't know which peer corresponds to the dest + // MAC, broadcast it. + router.PacketLogging.LogPacket("Broadcasting", key) + return router.relayBroadcast(router.Ourself.Peer, key) + default: + router.PacketLogging.LogPacket("Forwarding", key) + return router.relay(ForwardPacketKey{ + PacketKey: key, + SrcPeer: router.Ourself.Peer, + DstPeer: dstPeer}) + } +} + +func (router *NetworkRouter) handleForwardedPacket(key ForwardPacketKey) FlowOp { + if key.DstPeer != router.Ourself.Peer { + // it's not for us, we're just relaying it + router.PacketLogging.LogForwardPacket("Relaying", key) + return router.relay(key) + } + + // 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) + + srcMac := net.HardwareAddr(key.SrcMAC[:]) + dstMac := net.HardwareAddr(key.DstMAC[:]) + + switch newSrcMac, conflictPeer := router.Macs.AddForced(srcMac, key.SrcPeer); { + case newSrcMac: + log.Print("Discovered remote MAC ", srcMac, " at ", key.SrcPeer) + + case conflictPeer != nil: + log.Print("Discovered remote MAC ", srcMac, " at ", key.SrcPeer, " (was at ", conflictPeer, ")") + + // We need to clear out any flows destined to the MAC + // that forward to the old peer. + router.Overlay.(NetworkOverlay).InvalidateRoutes() + } + + router.PacketLogging.LogForwardPacket("Injecting", key) + injectFop := router.Bridge.InjectPacket(key.PacketKey) + dstPeer := router.Macs.Lookup(dstMac) + if dstPeer == router.Ourself.Peer { + return injectFop + } + + router.PacketLogging.LogForwardPacket("Relaying broadcast", key) + relayFop := router.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 + } +} + +// Routing + +func (router *NetworkRouter) relay(key ForwardPacketKey) FlowOp { + relayPeerName, found := 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:", key.DstPeer) + return DiscardingFlowOp{} + } + + conn, found := router.Ourself.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 DiscardingFlowOp{} + } + + return conn.(*LocalConnection).OverlayConn.(OverlayForwarder).Forward(key) +} + +func (router *NetworkRouter) relayBroadcast(srcPeer *Peer, key PacketKey) FlowOp { + nextHops := router.Routes.Broadcast(srcPeer.Name) + if len(nextHops) == 0 { + return DiscardingFlowOp{} + } + + op := NewMultiFlowOp(true) + + for _, conn := range router.Ourself.ConnectionsTo(nextHops) { + op.Add(conn.(*LocalConnection).OverlayConn.(OverlayForwarder).Forward(ForwardPacketKey{ + PacketKey: key, + SrcPeer: srcPeer, + DstPeer: conn.Remote()})) + } + + return op +} diff --git a/router/network_router_status.go b/router/network_router_status.go new file mode 100644 index 00000000..1b56d3e6 --- /dev/null +++ b/router/network_router_status.go @@ -0,0 +1,43 @@ +package router + +import ( + "time" +) + +type NetworkRouterStatus struct { + *Status + Interface string + CaptureStats map[string]int + MACs []MACStatus +} + +type MACStatus struct { + Mac string + Name string + NickName string + LastSeen time.Time +} + +func NewNetworkRouterStatus(router *NetworkRouter) *NetworkRouterStatus { + return &NetworkRouterStatus{ + NewStatus(router.Router), + router.Bridge.String(), + router.Bridge.Stats(), + NewMACStatusSlice(router.Macs)} +} + +func NewMACStatusSlice(cache *MacCache) []MACStatus { + cache.RLock() + defer cache.RUnlock() + + var slice []MACStatus + for key, entry := range cache.table { + slice = append(slice, MACStatus{ + intmac(key).String(), + entry.peer.Name.String(), + entry.peer.NickName, + entry.lastSeen}) + } + + return slice +} diff --git a/router/overlay.go b/router/overlay.go index e0e95364..13fd0428 100644 --- a/router/overlay.go +++ b/router/overlay.go @@ -4,34 +4,24 @@ import ( "net" ) -// Interface to overlay network packet handling type Overlay interface { - // Start consuming forwarded packets. - StartConsumingPackets(*Peer, *Peers, OverlayConsumer) error - - // Form a packet-forwarding connection. - MakeForwarder(ForwarderParams) (OverlayForwarder, error) - - // The routes have changed, so any cached information should - // be discarded. - InvalidateRoutes() - - // A mapping of a short id to a peer has changed - InvalidateShortIDs() - // Enhance a features map with overlay-related features AddFeaturesTo(map[string]string) + // Prepare on overlay connection. The connection should remain + // passive until it has been Confirm()ed. + PrepareConnection(OverlayConnectionParams) (OverlayConnection, error) + // Obtain diagnostic information specific to the overlay Diagnostics() interface{} } -type ForwarderParams struct { +type OverlayConnectionParams struct { RemotePeer *Peer // The local address of the corresponding TCP connection. Used to // derive the local IP address for sending. May differ for - // different forwarders. + // different overlay connections. LocalAddr *net.TCPAddr // The remote address of the corresponding TCP connection. Used to @@ -50,37 +40,29 @@ type ForwarderParams struct { SessionKey *[32]byte // Function to send a control message to the counterpart - // forwarder. + // overlay connection. SendControlMessage func(tag byte, msg []byte) error // Features passed at connection initiation Features map[string]string } -// When a consumer is called, the decoder will already have been used -// to decode the frame. -type OverlayConsumer func(ForwardPacketKey) FlowOp - -// All of the machinery to forward packets to a particular peer -type OverlayForwarder interface { - // Forward a packet across the connection. May be called as - // soon as the forwarder is created, in particular before - // Confirm(). The return value nil means the key could not be - // handled by this forwarder. - Forward(ForwardPacketKey) FlowOp - +// All of the machinery to manage overlay connectivity to a particular +// peer +type OverlayConnection interface { // Confirm that the connection is really wanted, and so the - // Overlay should begin heartbeats etc. to verify the - // operation of the forwarder. + // Overlay should begin heartbeats etc. to verify the operation of + // the overlay connection. Confirm() - // A channel indicating that the forwarder is established, - // i.e. its operation has been confirmed. + // A channel indicating that the overlay connection is + // established, i.e. its operation has been confirmed. EstablishedChannel() <-chan struct{} - // A channel indicating an error from the forwarder. The - // forwarder is not expected to be operational after the first - // error, so the channel only needs to buffer a single error. + // A channel indicating an error from the overlay connection. The + // overlay connection is not expected to be operational after the + // first error, so the channel only needs to buffer a single + // error. ErrorChannel() <-chan error Stop() @@ -96,20 +78,16 @@ type OverlayForwarder interface { type NullOverlay struct{} -func (NullOverlay) StartConsumingPackets(*Peer, *Peers, OverlayConsumer) error { - return nil +func (NullOverlay) AddFeaturesTo(map[string]string) { } -func (NullOverlay) MakeForwarder(ForwarderParams) (OverlayForwarder, error) { +func (NullOverlay) PrepareConnection(OverlayConnectionParams) (OverlayConnection, error) { return NullOverlay{}, nil } -func (NullOverlay) InvalidateRoutes() { +func (NullOverlay) Diagnostics() interface{} { + return nil } - -func (NullOverlay) InvalidateShortIDs() { -} - func (NullOverlay) Confirm() { } @@ -121,13 +99,6 @@ func (NullOverlay) ErrorChannel() <-chan error { return nil } -func (NullOverlay) AddFeaturesTo(map[string]string) { -} - -func (NullOverlay) Forward(ForwardPacketKey) FlowOp { - return DiscardingFlowOp{} -} - func (NullOverlay) Stop() { } @@ -137,7 +108,3 @@ func (NullOverlay) ControlMessage(byte, []byte) { func (NullOverlay) DisplayName() string { return "null" } - -func (NullOverlay) Diagnostics() interface{} { - return nil -} diff --git a/router/overlay_switch.go b/router/overlay_switch.go index aa18d174..9ead7987 100644 --- a/router/overlay_switch.go +++ b/router/overlay_switch.go @@ -13,16 +13,16 @@ import ( // uses the best one that seems to be working. type OverlaySwitch struct { - overlays map[string]Overlay + overlays map[string]NetworkOverlay overlayNames []string - compatOverlay Overlay + compatOverlay NetworkOverlay } func NewOverlaySwitch() *OverlaySwitch { - return &OverlaySwitch{overlays: make(map[string]Overlay)} + return &OverlaySwitch{overlays: make(map[string]NetworkOverlay)} } -func (osw *OverlaySwitch) Add(name string, overlay Overlay) { +func (osw *OverlaySwitch) Add(name string, overlay NetworkOverlay) { // check for repeated names if _, present := osw.overlays[name]; present { log.Fatal("OverlaySwitch: repeated overlay name") @@ -32,7 +32,7 @@ func (osw *OverlaySwitch) Add(name string, overlay Overlay) { osw.overlayNames = append(osw.overlayNames, name) } -func (osw *OverlaySwitch) SetCompatOverlay(overlay Overlay) { +func (osw *OverlaySwitch) SetCompatOverlay(overlay NetworkOverlay) { osw.compatOverlay = overlay } @@ -70,19 +70,19 @@ func (osw *OverlaySwitch) StartConsumingPackets(localPeer *Peer, peers *Peers, c } type namedOverlay struct { - Overlay + NetworkOverlay name string } // Find the common set of overlays supported by both sides, with the // ordering being the same on both sides too. -func (osw *OverlaySwitch) commonOverlays(params ForwarderParams) ([]namedOverlay, error) { +func (osw *OverlaySwitch) commonOverlays(params OverlayConnectionParams) ([]namedOverlay, error) { var peerOverlays []string if overlaysFeature, present := params.Features["Overlays"]; present { peerOverlays = strings.Split(overlaysFeature, " ") } - common := make(map[string]Overlay) + common := make(map[string]NetworkOverlay) for _, name := range peerOverlays { if overlay := osw.overlays[name]; overlay != nil { common[name] = overlay @@ -160,9 +160,9 @@ type subForwarderEvent struct { err error } -func (osw *OverlaySwitch) MakeForwarder(params ForwarderParams) (OverlayForwarder, error) { +func (osw *OverlaySwitch) PrepareConnection(params OverlayConnectionParams) (OverlayConnection, error) { if _, present := params.Features["Overlays"]; !present && osw.compatOverlay != nil { - return osw.compatOverlay.MakeForwarder(params) + return osw.compatOverlay.PrepareConnection(params) } overlays, err := osw.commonOverlays(params) @@ -200,11 +200,12 @@ func (osw *OverlaySwitch) MakeForwarder(params ForwarderParams) (OverlayForwarde return origSendControlMessage(ProtocolOverlayControlMsg, xmsg) } - subFwd, err := overlay.MakeForwarder(params) + subConn, err := overlay.PrepareConnection(params) if err != nil { fwd.stopFrom(0) return nil, err } + subFwd := subConn.(OverlayForwarder) subStopChan := make(chan struct{}) go monitorForwarder(i, eventsChan, subStopChan, subFwd) diff --git a/router/router.go b/router/router.go index 6d70eb92..d96b6fe2 100644 --- a/router/router.go +++ b/router/router.go @@ -11,28 +11,19 @@ import ( ) const ( - Port = 6783 - MaxUDPPacketSize = 65535 - ChannelSize = 16 - TCPHeartbeat = 30 * time.Second - GossipInterval = 30 * time.Second - MaxDuration = time.Duration(math.MaxInt64) - FastHeartbeat = 500 * time.Millisecond - SlowHeartbeat = 10 * time.Second - MaxMissedHeartbeats = 6 - HeartbeatTimeout = MaxMissedHeartbeats * SlowHeartbeat + Port = 6783 + ChannelSize = 16 + TCPHeartbeat = 30 * time.Second + GossipInterval = 30 * time.Second + MaxDuration = time.Duration(math.MaxInt64) - macMaxAge = 10 * time.Minute // [1] - acceptMaxTokens = 100 // [2] - acceptTokenDelay = 100 * time.Millisecond // [3] + acceptMaxTokens = 100 // [1] + acceptTokenDelay = 100 * time.Millisecond // [2] ) -// [1] should be greater than typical ARP cache expiries, i.e. > 3/2 * -// /proc/sys/net/ipv4_neigh/*/base_reachable_time_ms on Linux +// [1] capacity of token bucket for rate limiting accepts -// [2] capacity of token bucket for rate limiting accepts - -// [3] control rate at which new tokens are added to the bucket +// [2] control rate at which new tokens are added to the bucket var ( log = common.Log @@ -46,21 +37,12 @@ type Config struct { Password []byte ConnLimit int PeerDiscovery bool - BufSz int - PacketLogging PacketLogging - Bridge Bridge -} - -type PacketLogging interface { - LogPacket(string, PacketKey) - LogForwardPacket(string, ForwardPacketKey) } type Router struct { Config Overlay Overlay Ourself *LocalPeer - Macs *MacCache Peers *Peers Routes *Routes ConnectionMaker *ConnectionMaker @@ -73,28 +55,17 @@ type Router struct { func NewRouter(config Config, name PeerName, nickName string, overlay Overlay) *Router { router := &Router{Config: config, gossipChannels: make(GossipChannels)} - if router.Bridge == nil { - router.Bridge = NullBridge{} - } - if overlay == nil { overlay = NullOverlay{} } router.Overlay = overlay router.Ourself = NewLocalPeer(name, nickName, router) - router.Macs = NewMacCache(macMaxAge, - func(mac net.HardwareAddr, peer *Peer) { - log.Println("Expired MAC", mac, "at", peer) - }) router.Peers = NewPeers(router.Ourself) router.Peers.OnGC(func(peer *Peer) { - router.Macs.Delete(peer) log.Println("Removed unreachable peer", peer) }) - router.Peers.OnInvalidateShortIDs(overlay.InvalidateShortIDs) router.Routes = NewRoutes(router.Ourself, router.Peers) - router.Routes.OnChange(overlay.InvalidateRoutes) router.ConnectionMaker = NewConnectionMaker(router.Ourself, router.Peers, router.Port, router.PeerDiscovery) router.TopologyGossip = router.NewGossip("topology", router) router.acceptLimiter = NewTokenBucket(acceptMaxTokens, acceptTokenDelay) @@ -102,13 +73,10 @@ func NewRouter(config Config, name PeerName, nickName string, overlay Overlay) * return router } -// Start listening for TCP connections, locally captured packets, and -// forwarded packets. This is separate from NewRouter so -// that gossipers can register before we start forming connections. +// Start listening for TCP connections. This is separate from +// NewRouter so that gossipers can register before we start forming +// connections. func (router *Router) Start() { - log.Println("Sniffing traffic on", router.Bridge) - checkFatal(router.Bridge.StartConsumingPackets(router.handleCapturedPacket)) - checkFatal(router.Overlay.StartConsumingPackets(router.Ourself.Peer, router.Peers, router.handleForwardedPacket)) router.listenTCP(router.Port) } @@ -121,51 +89,6 @@ func (router *Router) UsingPassword() bool { return router.Password != nil } -func (router *Router) handleCapturedPacket(key PacketKey) FlowOp { - router.PacketLogging.LogPacket("Captured", key) - srcMac := net.HardwareAddr(key.SrcMAC[:]) - - switch newSrcMac, conflictPeer := router.Macs.Add(srcMac, router.Ourself.Peer); { - case newSrcMac: - log.Println("Discovered local MAC", srcMac) - - case conflictPeer != nil: - // The MAC cache has an entry for the source MAC - // associated with another peer. This probably means - // we are seeing a frame we injected ourself. That - // shouldn't happen, but discard it just in case. - log.Error("Captured frame from MAC (", srcMac, ") associated with another peer ", conflictPeer) - return DiscardingFlowOp{} - } - - // Discard STP broadcasts - if key.DstMAC == [...]byte{0x01, 0x80, 0xC2, 0x00, 0x00, 0x00} { - return DiscardingFlowOp{} - } - - 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 DiscardingFlowOp{} - case nil: - // If we don't know which peer corresponds to the dest - // MAC, broadcast it. - router.PacketLogging.LogPacket("Broadcasting", key) - return router.relayBroadcast(router.Ourself.Peer, key) - default: - router.PacketLogging.LogPacket("Forwarding", key) - return router.relay(ForwardPacketKey{ - PacketKey: key, - SrcPeer: router.Ourself.Peer, - DstPeer: dstPeer}) - } -} - func (router *Router) listenTCP(localPort int) { localAddr, err := net.ResolveTCPAddr("tcp4", fmt.Sprint(":", localPort)) checkFatal(err) @@ -195,95 +118,6 @@ func (router *Router) acceptTCP(tcpConn *net.TCPConn) { StartLocalConnection(connRemote, tcpConn, router, true) } -func (router *Router) handleForwardedPacket(key ForwardPacketKey) FlowOp { - if key.DstPeer != router.Ourself.Peer { - // it's not for us, we're just relaying it - router.PacketLogging.LogForwardPacket("Relaying", key) - return router.relay(key) - } - - // 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) - - srcMac := net.HardwareAddr(key.SrcMAC[:]) - dstMac := net.HardwareAddr(key.DstMAC[:]) - - switch newSrcMac, conflictPeer := router.Macs.AddForced(srcMac, key.SrcPeer); { - case newSrcMac: - log.Print("Discovered remote MAC ", srcMac, " at ", key.SrcPeer) - - case conflictPeer != nil: - log.Print("Discovered remote MAC ", srcMac, " at ", key.SrcPeer, " (was at ", conflictPeer, ")") - - // We need to clear out any flows destined to the MAC - // that forward to the old peer. - router.Overlay.InvalidateRoutes() - } - - router.PacketLogging.LogForwardPacket("Injecting", key) - injectFop := router.Bridge.InjectPacket(key.PacketKey) - dstPeer := router.Macs.Lookup(dstMac) - if dstPeer == router.Ourself.Peer { - return injectFop - } - - router.PacketLogging.LogForwardPacket("Relaying broadcast", key) - relayFop := router.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 - } -} - -// Routing - -func (router *Router) relay(key ForwardPacketKey) FlowOp { - relayPeerName, found := 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:", key.DstPeer) - return DiscardingFlowOp{} - } - - conn, found := router.Ourself.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 DiscardingFlowOp{} - } - - return conn.(*LocalConnection).forwarder.Forward(key) -} - -func (router *Router) relayBroadcast(srcPeer *Peer, key PacketKey) FlowOp { - nextHops := router.Routes.Broadcast(srcPeer.Name) - if len(nextHops) == 0 { - return DiscardingFlowOp{} - } - - op := NewMultiFlowOp(true) - - for _, conn := range router.Ourself.ConnectionsTo(nextHops) { - op.Add(conn.(*LocalConnection).forwarder.Forward(ForwardPacketKey{ - PacketKey: key, - SrcPeer: srcPeer, - DstPeer: conn.Remote()})) - } - - return op -} - // Gossiper methods - the Router is the topology Gossiper type TopologyGossipData struct { diff --git a/router/sleeve.go b/router/sleeve.go index cc6694ec..3b329a1a 100644 --- a/router/sleeve.go +++ b/router/sleeve.go @@ -71,7 +71,7 @@ type SleeveOverlay struct { forwarders map[PeerName]*sleeveForwarder } -func NewSleeveOverlay(localPort int) Overlay { +func NewSleeveOverlay(localPort int) NetworkOverlay { return &SleeveOverlay{localPort: localPort} } @@ -349,7 +349,7 @@ type controlMessage struct { msg []byte } -func (sleeve *SleeveOverlay) MakeForwarder(params ForwarderParams) (OverlayForwarder, error) { +func (sleeve *SleeveOverlay) PrepareConnection(params OverlayConnectionParams) (OverlayConnection, error) { name := sleeve.localPeer.NameByte var crypto sleeveCrypto if params.SessionKey != nil { diff --git a/router/status.go b/router/status.go index 469f9329..bb418d65 100644 --- a/router/status.go +++ b/router/status.go @@ -2,7 +2,6 @@ package router import ( "fmt" - "time" ) type Status struct { @@ -14,9 +13,6 @@ type Status struct { Name string NickName string Port int - Interface string - CaptureStats map[string]int - MACs []MACStatus Peers []PeerStatus UnicastRoutes []UnicastRouteStatus BroadcastRoutes []BroadcastRouteStatus @@ -25,13 +21,6 @@ type Status struct { OverlayDiagnostics interface{} } -type MACStatus struct { - Mac string - Name string - NickName string - LastSeen time.Time -} - type PeerStatus struct { Name string NickName string @@ -75,9 +64,6 @@ func NewStatus(router *Router) *Status { router.Ourself.Name.String(), router.Ourself.NickName, router.Port, - router.Bridge.String(), - router.Bridge.Stats(), - NewMACStatusSlice(router.Macs), NewPeerStatusSlice(router.Peers), NewUnicastRouteStatusSlice(router.Routes), NewBroadcastRouteStatusSlice(router.Routes), @@ -86,22 +72,6 @@ func NewStatus(router *Router) *Status { router.Overlay.Diagnostics()} } -func NewMACStatusSlice(cache *MacCache) []MACStatus { - cache.RLock() - defer cache.RUnlock() - - var slice []MACStatus - for key, entry := range cache.table { - slice = append(slice, MACStatus{ - intmac(key).String(), - entry.peer.Name.String(), - entry.peer.NickName, - entry.lastSeen}) - } - - return slice -} - func NewPeerStatusSlice(peers *Peers) []PeerStatus { var slice []PeerStatus @@ -176,7 +146,7 @@ func NewLocalConnectionStatusSlice(cm *ConnectionMaker) []LocalConnectionStatus state = "established" } lc, _ := conn.(*LocalConnection) - info := fmt.Sprintf("%-6v %v", lc.forwarder.DisplayName(), conn.Remote()) + info := fmt.Sprintf("%-6v %v", lc.OverlayConn.DisplayName(), conn.Remote()) slice = append(slice, LocalConnectionStatus{conn.RemoteTCPAddr(), conn.Outbound(), state, info}) } for address, target := range cm.targets {