From c2d6e9e8f1603f41c1aebaad5a6ec7023e34edc7 Mon Sep 17 00:00:00 2001 From: Arceliar Date: Sun, 24 May 2020 14:09:06 -0500 Subject: [PATCH 1/7] close listener when a multicast interface is removed --- src/multicast/multicast.go | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/multicast/multicast.go b/src/multicast/multicast.go index fc7b137..ad097b8 100644 --- a/src/multicast/multicast.go +++ b/src/multicast/multicast.go @@ -136,6 +136,7 @@ func (m *Multicast) _stop() error { m.log.Infoln("Stopping multicast module") m.isOpen = false for name := range m.listeners { + m.listeners[name].listener.Listener.Close() close(m.listeners[name].stop) delete(m.listeners, name) } @@ -213,6 +214,7 @@ func (m *Multicast) _monitorInterfaceChanges() { for name, intf := range m.listeners { if _, ok := m._interfaces[name]; !ok { // This is a disappeared interface. Stop the announcer. + intf.listener.Listener.Close() close(intf.stop) delete(m.listeners, name) m.log.Debugln("Stopped multicasting on", name) From 1e471e3712d14d7434b14e6d21b99be169bfaec5 Mon Sep 17 00:00:00 2001 From: Arceliar Date: Sun, 24 May 2020 14:43:38 -0500 Subject: [PATCH 2/7] back to master's version of multicast, lets try rewriting it again --- src/multicast/multicast.go | 292 ++++++++++++++++++------------------- 1 file changed, 140 insertions(+), 152 deletions(-) diff --git a/src/multicast/multicast.go b/src/multicast/multicast.go index ad097b8..30f6615 100644 --- a/src/multicast/multicast.go +++ b/src/multicast/multicast.go @@ -15,57 +15,38 @@ import ( "golang.org/x/net/ipv6" ) -const ( - // GroupAddr contains the multicast group and port used for multicast packets. - GroupAddr = "[ff02::114]:9001" -) - // Multicast represents the multicast advertisement and discovery mechanism used // by Yggdrasil to find peers on the same subnet. When a beacon is received on a // configured multicast interface, Yggdrasil will attempt to peer with that node // automatically. type Multicast struct { phony.Inbox - core *yggdrasil.Core - config *config.NodeState - log *log.Logger - sock *ipv6.PacketConn - groupAddr *net.UDPAddr - listeners map[string]*multicastInterface - listenPort uint16 - isOpen bool - monitor *time.Timer - platformhandler *time.Timer - _interfaces map[string]net.Interface - _interfaceAddrs map[string]addrInfo + core *yggdrasil.Core + config *config.NodeState + log *log.Logger + sock *ipv6.PacketConn + groupAddr string + listeners map[string]*listenerInfo + listenPort uint16 + isOpen bool } -type addrInfo struct { - addrs []net.Addr - time time.Time -} - -type multicastInterface struct { - phony.Inbox - sock *ipv6.PacketConn - destAddr net.UDPAddr +type listenerInfo struct { listener *yggdrasil.TcpListener - zone string - timer *time.Timer + time time.Time interval time.Duration - stop chan interface{} } // Init prepares the multicast interface for use. -func (m *Multicast) Init(core *yggdrasil.Core, state *config.NodeState, log *log.Logger, options interface{}) (err error) { +func (m *Multicast) Init(core *yggdrasil.Core, state *config.NodeState, log *log.Logger, options interface{}) error { m.core = core m.config = state m.log = log - m.listeners = make(map[string]*multicastInterface) + m.listeners = make(map[string]*listenerInfo) current := m.config.GetCurrent() m.listenPort = current.LinkLocalTCPPort - m.groupAddr, err = net.ResolveUDPAddr("udp6", GroupAddr) - return + m.groupAddr = "[ff02::114]:9001" + return nil } // Start starts the multicast interface. This launches goroutines which will @@ -88,7 +69,7 @@ func (m *Multicast) _start() error { return nil } m.log.Infoln("Starting multicast module") - addr, err := net.ResolveUDPAddr("udp", GroupAddr) + addr, err := net.ResolveUDPAddr("udp", m.groupAddr) if err != nil { return err } @@ -108,7 +89,7 @@ func (m *Multicast) _start() error { m.isOpen = true go m.listen() m.Act(nil, m._multicastStarted) - m.Act(nil, m._monitorInterfaceChanges) + m.Act(nil, m._announce) return nil } @@ -135,14 +116,6 @@ func (m *Multicast) Stop() error { func (m *Multicast) _stop() error { m.log.Infoln("Stopping multicast module") m.isOpen = false - for name := range m.listeners { - m.listeners[name].listener.Listener.Close() - close(m.listeners[name].stop) - delete(m.listeners, name) - } - if m.platformhandler != nil { - m.platformhandler.Stop() - } if m.sock != nil { m.sock.Close() } @@ -175,93 +148,10 @@ func (m *Multicast) _updateConfig(config *config.NodeConfig) { m.log.Debugln("Reloaded multicast configuration successfully") } -func (m *Multicast) _monitorInterfaceChanges() { - m._updateInterfaces() // update interfaces and interfaceAddrs - - // Look for interfaces we don't know about yet. - for name, intf := range m._interfaces { - if _, ok := m.listeners[name]; !ok { - // Look up interface addresses. - addrs := m._interfaceAddrs[intf.Name].addrs - // Find the first link-local address. - for _, addr := range addrs { - addrIP, _, _ := net.ParseCIDR(addr.String()) - // Join the multicast group. - m.sock.JoinGroup(&intf, m.groupAddr) - // Construct a listener on this address. - listenaddr := fmt.Sprintf("[%s%%%s]:%d", addrIP, intf.Name, m.listenPort) - listener, err := m.core.ListenTCP(listenaddr) - if err != nil { - m.log.Warnln("Not multicasting on", name, "due to error:", err) - continue - } - // This is a new interface. Start an announcer for it. - multicastInterface := &multicastInterface{ - sock: m.sock, - destAddr: *m.groupAddr, - listener: listener, - stop: make(chan interface{}), - zone: name, - } - multicastInterface.Act(m, multicastInterface._announce) - m.listeners[name] = multicastInterface - m.log.Debugln("Started multicasting on", name) - break - } - } - } - // Look for interfaces we knew about but are no longer there. - for name, intf := range m.listeners { - if _, ok := m._interfaces[name]; !ok { - // This is a disappeared interface. Stop the announcer. - intf.listener.Listener.Close() - close(intf.stop) - delete(m.listeners, name) - m.log.Debugln("Stopped multicasting on", name) - } - } - // Queue the next check. - m.monitor = time.AfterFunc(time.Second, func() { - m.Act(nil, m._monitorInterfaceChanges) - }) -} - -func (m *multicastInterface) _announce() { - // Check if the multicast interface has been stopped. This will happen - // if it disappears from the system or goes down. - select { - case <-m.stop: - return - default: - } - // Send the beacon. - lladdr := m.listener.Listener.Addr().String() - if a, err := net.ResolveTCPAddr("tcp6", lladdr); err == nil { - a.Zone = "" - msg := []byte(a.String()) - m.sock.WriteTo(msg, nil, &m.destAddr) - } - // Queue the next beacon. - if m.interval.Seconds() < 15 { - m.interval += time.Second - } - m.timer = time.AfterFunc(m.interval, func() { - m.Act(nil, m._announce) - }) -} - // GetInterfaces returns the currently known/enabled multicast interfaces. It is // expected that UpdateInterfaces has been called at least once before calling // this method. func (m *Multicast) Interfaces() map[string]net.Interface { - var interfaces map[string]net.Interface - phony.Block(m, func() { - interfaces = m._interfaces - }) - return interfaces -} - -func (m *Multicast) _updateInterfaces() { interfaces := make(map[string]net.Interface) // Get interface expressions from config current := m.config.GetCurrent() @@ -272,7 +162,6 @@ func (m *Multicast) _updateInterfaces() { panic(err) } // Work out which interfaces to announce on - interfaceAddrs := make(map[string]addrInfo) for _, iface := range allifaces { if iface.Flags&net.FlagUp == 0 { // Ignore interfaces that are down @@ -286,26 +175,6 @@ func (m *Multicast) _updateInterfaces() { // Ignore point-to-point interfaces continue } - var aInfo addrInfo - var isIn bool - if aInfo, isIn = m._interfaceAddrs[iface.Name]; isIn && time.Since(aInfo.time) < time.Minute { - // don't call iface.Addrs, it's unlikely things have changed - } else { - aInfo.addrs, _ = iface.Addrs() - aInfo.time = time.Now() - } - lladdrs := aInfo.addrs[:0] - for _, addr := range aInfo.addrs { - addrIP, _, _ := net.ParseCIDR(addr.String()) - if addrIP.To4() == nil && addrIP.IsLinkLocalUnicast() { - lladdrs = append(lladdrs, addr) - } - } - aInfo.addrs = lladdrs - if len(lladdrs) == 0 { - // Ignore interfaces without link-local addresses - continue - } for _, expr := range exprs { // Compile each regular expression e, err := regexp.Compile(expr) @@ -315,16 +184,136 @@ func (m *Multicast) _updateInterfaces() { // Does the interface match the regular expression? Store it if so if e.MatchString(iface.Name) { interfaces[iface.Name] = iface - interfaceAddrs[iface.Name] = aInfo } } } - m._interfaces = interfaces - m._interfaceAddrs = interfaceAddrs + return interfaces +} + +func (m *Multicast) _announce() { + if !m.isOpen { + return + } + groupAddr, err := net.ResolveUDPAddr("udp6", m.groupAddr) + if err != nil { + panic(err) + } + destAddr, err := net.ResolveUDPAddr("udp6", m.groupAddr) + if err != nil { + panic(err) + } + interfaces := m.Interfaces() + // There might be interfaces that we configured listeners for but are no + // longer up - if that's the case then we should stop the listeners + for name, info := range m.listeners { + // Prepare our stop function! + stop := func() { + info.listener.Stop() + delete(m.listeners, name) + m.log.Debugln("No longer multicasting on", name) + } + // If the interface is no longer visible on the system then stop the + // listener, as another one will be started further down + if _, ok := interfaces[name]; !ok { + stop() + continue + } + // It's possible that the link-local listener address has changed so if + // that is the case then we should clean up the interface listener + found := false + listenaddr, err := net.ResolveTCPAddr("tcp6", info.listener.Listener.Addr().String()) + if err != nil { + stop() + continue + } + // Find the interface that matches the listener + if intf, err := net.InterfaceByName(name); err == nil { + if addrs, err := intf.Addrs(); err == nil { + // Loop through the addresses attached to that listener and see if any + // of them match the current address of the listener + for _, addr := range addrs { + if ip, _, err := net.ParseCIDR(addr.String()); err == nil { + // Does the interface address match our listener address? + if ip.Equal(listenaddr.IP) { + found = true + break + } + } + } + } + } + // If the address has not been found on the adapter then we should stop + // and clean up the TCP listener. A new one will be created below if a + // suitable link-local address is found + if !found { + stop() + } + } + // Now that we have a list of valid interfaces from the operating system, + // we can start checking if we can send multicasts on them + for _, iface := range interfaces { + // Find interface addresses + addrs, err := iface.Addrs() + if err != nil { + panic(err) + } + for _, addr := range addrs { + addrIP, _, _ := net.ParseCIDR(addr.String()) + // Ignore IPv4 addresses + if addrIP.To4() != nil { + continue + } + // Ignore non-link-local addresses + if !addrIP.IsLinkLocalUnicast() { + continue + } + // Join the multicast group + m.sock.JoinGroup(&iface, groupAddr) + // Try and see if we already have a TCP listener for this interface + var info *listenerInfo + if nfo, ok := m.listeners[iface.Name]; !ok || nfo.listener.Listener == nil { + // No listener was found - let's create one + listenaddr := fmt.Sprintf("[%s%%%s]:%d", addrIP, iface.Name, m.listenPort) + if li, err := m.core.ListenTCP(listenaddr); err == nil { + m.log.Debugln("Started multicasting on", iface.Name) + // Store the listener so that we can stop it later if needed + info = &listenerInfo{listener: li, time: time.Now()} + m.listeners[iface.Name] = info + } else { + m.log.Warnln("Not multicasting on", iface.Name, "due to error:", err) + } + } else { + // An existing listener was found + info = m.listeners[iface.Name] + } + // Make sure nothing above failed for some reason + if info == nil { + continue + } + if time.Since(info.time) < info.interval { + continue + } + // Get the listener details and construct the multicast beacon + lladdr := info.listener.Listener.Addr().String() + if a, err := net.ResolveTCPAddr("tcp6", lladdr); err == nil { + a.Zone = "" + destAddr.Zone = iface.Name + msg := []byte(a.String()) + m.sock.WriteTo(msg, nil, destAddr) + } + if info.interval.Seconds() < 15 { + info.interval += time.Second + } + break + } + } + time.AfterFunc(time.Second, func() { + m.Act(nil, m._announce) + }) } func (m *Multicast) listen() { - groupAddr, err := net.ResolveUDPAddr("udp6", GroupAddr) + groupAddr, err := net.ResolveUDPAddr("udp6", m.groupAddr) if err != nil { panic(err) } @@ -357,7 +346,6 @@ func (m *Multicast) listen() { if addr.IP.String() != from.IP.String() { continue } - // Note that m.Interfaces would block if it was being run by the actor itself if _, ok := m.Interfaces()[from.Zone]; ok { addr.Zone = "" if err := m.core.CallPeer("tcp://"+addr.String(), from.Zone); err != nil { From 98816f34b233bbe0a9d44fa8d3050fd2b9052606 Mon Sep 17 00:00:00 2001 From: Arceliar Date: Sun, 24 May 2020 15:24:39 -0500 Subject: [PATCH 3/7] don't spam calls to net.Interfaces and net.Interface.Addrs (hopefully) --- src/multicast/multicast.go | 74 +++++++++++++++++++++++++++----------- 1 file changed, 53 insertions(+), 21 deletions(-) diff --git a/src/multicast/multicast.go b/src/multicast/multicast.go index 30f6615..7de3cfa 100644 --- a/src/multicast/multicast.go +++ b/src/multicast/multicast.go @@ -21,14 +21,20 @@ import ( // automatically. type Multicast struct { phony.Inbox - core *yggdrasil.Core - config *config.NodeState - log *log.Logger - sock *ipv6.PacketConn - groupAddr string - listeners map[string]*listenerInfo - listenPort uint16 - isOpen bool + core *yggdrasil.Core + config *config.NodeState + log *log.Logger + sock *ipv6.PacketConn + groupAddr string + listeners map[string]*listenerInfo + listenPort uint16 + isOpen bool + _interfaces map[string]interfaceInfo +} + +type interfaceInfo struct { + iface net.Interface + addrs []net.Addr } type listenerInfo struct { @@ -43,6 +49,7 @@ func (m *Multicast) Init(core *yggdrasil.Core, state *config.NodeState, log *log m.config = state m.log = log m.listeners = make(map[string]*listenerInfo) + m._interfaces = make(map[string]interfaceInfo) current := m.config.GetCurrent() m.listenPort = current.LinkLocalTCPPort m.groupAddr = "[ff02::114]:9001" @@ -148,10 +155,35 @@ func (m *Multicast) _updateConfig(config *config.NodeConfig) { m.log.Debugln("Reloaded multicast configuration successfully") } -// GetInterfaces returns the currently known/enabled multicast interfaces. It is -// expected that UpdateInterfaces has been called at least once before calling -// this method. +func (m *Multicast) _updateInterfaces() { + interfaces := make(map[string]interfaceInfo) + intfs := m.getAllowedInterfaces() + for _, intf := range intfs { + addrs, err := intf.Addrs() + if err != nil { + m.log.Warnf("Failed up get addresses for interface %s: %s", intf.Name, err) + continue + } + interfaces[intf.Name] = interfaceInfo{ + iface: intf, + addrs: addrs, + } + } + m._interfaces = interfaces +} + func (m *Multicast) Interfaces() map[string]net.Interface { + interfaces := make(map[string]net.Interface) + phony.Block(m, func() { + for _, info := range m._interfaces { + interfaces[info.iface.Name] = info.iface + } + }) + return interfaces +} + +// getAllowedInterfaces returns the currently known/enabled multicast interfaces. +func (m *Multicast) getAllowedInterfaces() map[string]net.Interface { interfaces := make(map[string]net.Interface) // Get interface expressions from config current := m.config.GetCurrent() @@ -194,6 +226,7 @@ func (m *Multicast) _announce() { if !m.isOpen { return } + m._updateInterfaces() groupAddr, err := net.ResolveUDPAddr("udp6", m.groupAddr) if err != nil { panic(err) @@ -202,7 +235,6 @@ func (m *Multicast) _announce() { if err != nil { panic(err) } - interfaces := m.Interfaces() // There might be interfaces that we configured listeners for but are no // longer up - if that's the case then we should stop the listeners for name, info := range m.listeners { @@ -214,7 +246,7 @@ func (m *Multicast) _announce() { } // If the interface is no longer visible on the system then stop the // listener, as another one will be started further down - if _, ok := interfaces[name]; !ok { + if _, ok := m._interfaces[name]; !ok { stop() continue } @@ -251,13 +283,9 @@ func (m *Multicast) _announce() { } // Now that we have a list of valid interfaces from the operating system, // we can start checking if we can send multicasts on them - for _, iface := range interfaces { - // Find interface addresses - addrs, err := iface.Addrs() - if err != nil { - panic(err) - } - for _, addr := range addrs { + for _, info := range m._interfaces { + iface := info.iface + for _, addr := range info.addrs { addrIP, _, _ := net.ParseCIDR(addr.String()) // Ignore IPv4 addresses if addrIP.To4() != nil { @@ -346,7 +374,11 @@ func (m *Multicast) listen() { if addr.IP.String() != from.IP.String() { continue } - if _, ok := m.Interfaces()[from.Zone]; ok { + var interfaces map[string]interfaceInfo + phony.Block(m, func() { + interfaces = m._interfaces + }) + if _, ok := interfaces[from.Zone]; ok { addr.Zone = "" if err := m.core.CallPeer("tcp://"+addr.String(), from.Zone); err != nil { m.log.Debugln("Call from multicast failed:", err) From 7778a47a8f224cf4e16cf8008fbed62d8ee18ceb Mon Sep 17 00:00:00 2001 From: Arceliar Date: Sun, 24 May 2020 15:46:18 -0500 Subject: [PATCH 4/7] fix darwin compile problem --- src/multicast/multicast_darwin.go | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/src/multicast/multicast_darwin.go b/src/multicast/multicast_darwin.go index 91cb5ae..ceff5b4 100644 --- a/src/multicast/multicast_darwin.go +++ b/src/multicast/multicast_darwin.go @@ -32,6 +32,9 @@ import ( var awdlGoroutineStarted bool func (m *Multicast) _multicastStarted() { + if !m.isOpen { + return + } C.StopAWDLBrowsing() for intf := range m._interfaces { if intf == "awdl0" { @@ -39,7 +42,7 @@ func (m *Multicast) _multicastStarted() { break } } - m.platformhandler = time.AfterFunc(time.Minute, func() { + time.AfterFunc(time.Minute, func() { m.Act(nil, m._multicastStarted) }) } From 40bfd207f56c26159322b7dbc6bd7a9b443a0c2e Mon Sep 17 00:00:00 2001 From: Arceliar Date: Mon, 25 May 2020 12:23:38 -0500 Subject: [PATCH 5/7] don't store every node we hear from in the DHT, only ones we already know about or that are important --- src/yggdrasil/dht.go | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/yggdrasil/dht.go b/src/yggdrasil/dht.go index 8efc549..0b951c9 100644 --- a/src/yggdrasil/dht.go +++ b/src/yggdrasil/dht.go @@ -260,7 +260,9 @@ func (t *dht) handleRes(res *dhtRes) { key: res.Key, coords: res.Coords, } - t.insert(&rinfo) + if _, isIn := t.table[*rinfo.getNodeID()]; isIn || t.isImportant(&rinfo) { + t.insert(&rinfo) + } for _, info := range res.Infos { if *info.getNodeID() == t.nodeID { continue From eefabb5f9fd128aa51350c9e81c6b771f2f32984 Mon Sep 17 00:00:00 2001 From: Arceliar Date: Mon, 25 May 2020 12:44:06 -0500 Subject: [PATCH 6/7] disregard nodes if they're unimportant, even if they're already in the DHT --- src/yggdrasil/dht.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/yggdrasil/dht.go b/src/yggdrasil/dht.go index 0b951c9..013fd1e 100644 --- a/src/yggdrasil/dht.go +++ b/src/yggdrasil/dht.go @@ -260,7 +260,7 @@ func (t *dht) handleRes(res *dhtRes) { key: res.Key, coords: res.Coords, } - if _, isIn := t.table[*rinfo.getNodeID()]; isIn || t.isImportant(&rinfo) { + if t.isImportant(&rinfo) { t.insert(&rinfo) } for _, info := range res.Infos { From 8cca565ac46a4e412ffe7c9089d1c4ecb86fb492 Mon Sep 17 00:00:00 2001 From: Neil Alexander Date: Mon, 25 May 2020 22:08:53 +0100 Subject: [PATCH 7/7] Update go.mod/go.sum for yggdrasil-extras for iOS builds --- go.mod | 1 + go.sum | 2 ++ 2 files changed, 3 insertions(+) diff --git a/go.mod b/go.mod index bf8e757..cea62b2 100644 --- a/go.mod +++ b/go.mod @@ -12,6 +12,7 @@ require ( github.com/mitchellh/mapstructure v1.1.2 github.com/vishvananda/netlink v1.0.0 github.com/vishvananda/netns v0.0.0-20190625233234-7109fa855b0f // indirect + github.com/yggdrasil-network/yggdrasil-extras v0.0.0-20200525205615-6c8a4a2e8855 // indirect golang.org/x/crypto v0.0.0-20200221231518-2aa609cf4a9d golang.org/x/net v0.0.0-20200301022130-244492dfa37a golang.org/x/sys v0.0.0-20200302150141-5c8b2ff67527 diff --git a/go.sum b/go.sum index f67b680..013abbe 100644 --- a/go.sum +++ b/go.sum @@ -31,6 +31,8 @@ github.com/vishvananda/netlink v1.0.0 h1:bqNY2lgheFIu1meHUFSH3d7vG93AFyqg3oGbJCO github.com/vishvananda/netlink v1.0.0/go.mod h1:+SR5DhBJrl6ZM7CoCKvpw5BKroDKQ+PJqOg65H/2ktk= github.com/vishvananda/netns v0.0.0-20190625233234-7109fa855b0f h1:nBX3nTcmxEtHSERBJaIo1Qa26VwRaopnZmfDQUXsF4I= github.com/vishvananda/netns v0.0.0-20190625233234-7109fa855b0f/go.mod h1:ZjcWmFBXmLKZu9Nxj3WKYEafiSqer2rnvPr0en9UNpI= +github.com/yggdrasil-network/yggdrasil-extras v0.0.0-20200525205615-6c8a4a2e8855 h1:xLQihK8bAKOEDii/Z39dHTgSJzetm2TQ1YKRPRX87R4= +github.com/yggdrasil-network/yggdrasil-extras v0.0.0-20200525205615-6c8a4a2e8855/go.mod h1:xQdsh08Io6nV4WRnOVTe6gI8/2iTvfLDQ0CYa5aMt+I= golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= golang.org/x/crypto v0.0.0-20191002192127-34f69633bfdc/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= golang.org/x/crypto v0.0.0-20200221231518-2aa609cf4a9d h1:1ZiEyfaQIg3Qh0EoqpwAakHVhecoE5wlSg5GjnafJGw=