Yggdrasil 0.5.8

Causes issues such as #1156.

CHANGELOG.md

@@ -26,6 +26,32 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 - in case of vulnerabilities.
 -->
 
+## [0.5.8] - 2024-08-12
+
+### Fixed
+
+* A bug which caused startup problems on Windows and FreeBSD should be fixed
+* Resolved some minor link state and listener management bugs during shutdown
+
+## [0.5.7] - 2024-08-05
+
+### Added
+
+* WebSocket support for peerings, by using the new `ws://` scheme in `Listen` and `Peers`
+  * Additionally, the `wss://` scheme can be used to connect to a WebSocket peer behind a HTTPS reverse proxy
+
+### Changed
+
+* On Linux, the TUN adapter now uses vectorised reads/writes where possible, which should reduce the amount of CPU time spent on syscalls and potentially improve throughput
+* Link error handling has been improved and various link error messages have been rewritten to be clearer
+* Upgrade dependencies
+
+### Fixed
+
+* Multiple multicast connections to the same remote machine should now work correctly
+  * You may get two connections in some cases, one inbound and one outbound, this is known and will not cause problems
+* Running as a Windows service should be more reliable with service startup and shutdown bugs fixed
+
 ## [0.5.6] - 2024-05-30
 
 * Go 1.21 is now required to build Yggdrasil

cmd/yggdrasil/main.go

@@ -53,13 +53,13 @@ func main() {
 	getpkey := flag.Bool("publickey", false, "use in combination with either -useconf or -useconffile, outputs your public key")
 	loglevel := flag.String("loglevel", "info", "loglevel to enable")
 	flag.Parse()
-
+	
+	done := make(chan struct{})
+	defer close(done)
+	
 	// Catch interrupts from the operating system to exit gracefully.
 	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
 
-	// Capture the service being stopped on Windows.
-	minwinsvc.SetOnExit(cancel)
-
 	// Create a new logger that logs output to stdout.
 	var logger *log.Logger
 	switch *logto {
@@ -271,6 +271,14 @@ func main() {
 			n.tun.SetupAdminHandlers(n.admin)
 		}
 	}
+	
+	//Windows service shutdown
+	minwinsvc.SetOnExit(func() {
+		logger.Infof("Shutting down service ...")
+		cancel()
+		// Wait for all parts to shutdown properly
+		<-done
+	})
 
 	// Block until we are told to shut down.
 	<-ctx.Done()

src/core/link.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"encoding/hex"
+	"errors"
 	"fmt"
 	"io"
 	"net"
@@ -40,7 +41,8 @@ type links struct {
 	ws    *linkWS    // WS interface support
 	wss   *linkWSS   // WSS interface support
 	// _links can only be modified safely from within the links actor
-	_links map[linkInfo]*link // *link is nil if connection in progress
+	_links     map[linkInfo]*link // *link is nil if connection in progress
+	_listeners map[*Listener]context.CancelFunc
 }
 
 type linkProtocol interface {
@@ -85,13 +87,6 @@ func (l *Listener) Addr() net.Addr {
 	return l.listener.Addr()
 }
 
-func (l *Listener) Close() error {
-	l.Cancel()
-	err := l.listener.Close()
-	<-l.ctx.Done()
-	return err
-}
-
 func (l *links) init(c *Core) error {
 	l.core = c
 	l.tcp = l.newLinkTCP()
@@ -102,32 +97,18 @@ func (l *links) init(c *Core) error {
 	l.ws = l.newLinkWS()
 	l.wss = l.newLinkWSS()
 	l._links = make(map[linkInfo]*link)
-
-	var listeners []ListenAddress
-	phony.Block(c, func() {
-		listeners = make([]ListenAddress, 0, len(c.config._listeners))
-		for listener := range c.config._listeners {
-			listeners = append(listeners, listener)
-		}
-	})
+	l._listeners = make(map[*Listener]context.CancelFunc)
 
 	return nil
 }
 
 func (l *links) shutdown() {
-	phony.Block(l.tcp, func() {
-		for l := range l.tcp._listeners {
-			_ = l.Close()
+	phony.Block(l, func() {
+		for listener := range l._listeners {
+			_ = listener.listener.Close()
 		}
-	})
-	phony.Block(l.tls, func() {
-		for l := range l.tls._listeners {
-			_ = l.Close()
-		}
-	})
-	phony.Block(l.unix, func() {
-		for l := range l.unix._listeners {
-			_ = l.Close()
+		for _, link := range l._links {
+			_ = link._conn.Close()
 		}
 	})
 }
@@ -140,7 +121,7 @@ const ErrLinkAlreadyConfigured = linkError("peer is already configured")
 const ErrLinkNotConfigured = linkError("peer is not configured")
 const ErrLinkPriorityInvalid = linkError("priority value is invalid")
 const ErrLinkPinnedKeyInvalid = linkError("pinned public key is invalid")
-const ErrLinkPasswordInvalid = linkError("password is invalid")
+const ErrLinkPasswordInvalid = linkError("invalid password supplied")
 const ErrLinkUnrecognisedSchema = linkError("link schema unknown")
 const ErrLinkMaxBackoffInvalid = linkError("max backoff duration invalid")
 
@@ -363,9 +344,11 @@ func (l *links) add(u *url.URL, sintf string, linkType linkType) error {
 				_ = lc.Close()
 				phony.Block(l, func() {
 					state._conn = nil
-					if state._err = err; state._err != nil {
-						state._errtime = time.Now()
+					if err == nil {
+						err = fmt.Errorf("remote side closed the connection")
 					}
+					state._err = err
+					state._errtime = time.Now()
 				})
 
 				// If the link is persistently configured, back off if needed
@@ -455,11 +438,18 @@ func (l *links) listen(u *url.URL, sintf string) (*Listener, error) {
 		options.password = []byte(p)
 	}
 
+	phony.Block(l, func() {
+		l._listeners[li] = cancel
+	})
+
 	go func() {
-		l.core.log.Infof("%s listener started on %s", strings.ToUpper(u.Scheme), listener.Addr())
-		defer l.core.log.Infof("%s listener stopped on %s", strings.ToUpper(u.Scheme), listener.Addr())
+		l.core.log.Infof("%s listener started on %s", strings.ToUpper(u.Scheme), li.listener.Addr())
+		defer l.core.log.Infof("%s listener stopped on %s", strings.ToUpper(u.Scheme), li.listener.Addr())
+		defer phony.Block(l, func() {
+			delete(l._listeners, li)
+		})
 		for {
-			conn, err := listener.Accept()
+			conn, err := li.listener.Accept()
 			if err != nil {
 				return
 			}
@@ -515,13 +505,22 @@ func (l *links) listen(u *url.URL, sintf string) (*Listener, error) {
 					// Store the state of the link so that it can be queried later.
 					l._links[info] = state
 				})
+				defer phony.Block(l, func() {
+					if l._links[info] == state {
+						delete(l._links, info)
+					}
+				})
 				if lc == nil {
 					return
 				}
 
 				// Give the connection to the handler. The handler will block
 				// for the lifetime of the connection.
-				if err = l.handler(linkTypeIncoming, options, lc, nil); err != nil && err != io.EOF {
+				switch err = l.handler(linkTypeIncoming, options, lc, nil); {
+				case err == nil:
+				case errors.Is(err, io.EOF):
+				case errors.Is(err, net.ErrClosed):
+				default:
 					l.core.log.Debugf("Link %s error: %s\n", u.Host, err)
 				}
 
@@ -529,11 +528,6 @@ func (l *links) listen(u *url.URL, sintf string) (*Listener, error) {
 				// try to close the underlying socket just in case and then
 				// drop the link state.
 				_ = lc.Close()
-				phony.Block(l, func() {
-					if l._links[info] == state {
-						delete(l._links, info)
-					}
-				})
 			}(conn)
 		}
 	}()
@@ -647,7 +641,7 @@ func (l *links) handler(linkType linkType, options linkOptions, conn net.Conn, s
 		l.core.log.Infof("Disconnected %s: %s, source %s; error: %s",
 			dir, remoteStr, localStr, err)
 	}
-	return nil
+	return err
 }
 
 func urlForLinkInfo(u url.URL) url.URL {

src/core/link_tcp.go

@@ -15,7 +15,6 @@ type linkTCP struct {
 	phony.Inbox
 	*links
 	listenconfig *net.ListenConfig
-	_listeners   map[*Listener]context.CancelFunc
 }
 
 func (l *links) newLinkTCP() *linkTCP {
@@ -24,7 +23,6 @@ func (l *links) newLinkTCP() *linkTCP {
 		listenconfig: &net.ListenConfig{
 			KeepAlive: -1,
 		},
-		_listeners: map[*Listener]context.CancelFunc{},
 	}
 	lt.listenconfig.Control = lt.tcpContext
 	return lt

src/core/link_tls.go

@@ -13,10 +13,9 @@ import (
 type linkTLS struct {
 	phony.Inbox
 	*links
-	tcp        *linkTCP
-	listener   *net.ListenConfig
-	config     *tls.Config
-	_listeners map[*Listener]context.CancelFunc
+	tcp      *linkTCP
+	listener *net.ListenConfig
+	config   *tls.Config
 }
 
 func (l *links) newLinkTLS(tcp *linkTCP) *linkTLS {
@@ -27,8 +26,7 @@ func (l *links) newLinkTLS(tcp *linkTCP) *linkTLS {
 			Control:   tcp.tcpContext,
 			KeepAlive: -1,
 		},
-		config:     l.core.config.tls.Clone(),
-		_listeners: map[*Listener]context.CancelFunc{},
+		config: l.core.config.tls.Clone(),
 	}
 	return lt
 }

src/core/link_unix.go

@@ -12,9 +12,8 @@ import (
 type linkUNIX struct {
 	phony.Inbox
 	*links
-	dialer     *net.Dialer
-	listener   *net.ListenConfig
-	_listeners map[*Listener]context.CancelFunc
+	dialer   *net.Dialer
+	listener *net.ListenConfig
 }
 
 func (l *links) newLinkUNIX() *linkUNIX {
@@ -27,7 +26,6 @@ func (l *links) newLinkUNIX() *linkUNIX {
 		listener: &net.ListenConfig{
 			KeepAlive: -1,
 		},
-		_listeners: map[*Listener]context.CancelFunc{},
 	}
 	return lt
 }

src/core/link_ws.go

@@ -14,6 +14,7 @@ import (
 type linkWS struct {
 	phony.Inbox
 	*links
+	listenconfig *net.ListenConfig
 }
 
 type linkWSConn struct {
@@ -78,6 +79,9 @@ func (s *wsServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 func (l *links) newLinkWS() *linkWS {
 	lt := &linkWS{
 		links: l,
+		listenconfig: &net.ListenConfig{
+			KeepAlive: -1,
+		},
 	}
 	return lt
 }
@@ -95,7 +99,7 @@ func (l *linkWS) dial(ctx context.Context, url *url.URL, info linkInfo, options
 }
 
 func (l *linkWS) listen(ctx context.Context, url *url.URL, _ string) (net.Listener, error) {
-	nl, err := net.Listen("tcp", url.Host)
+	nl, err := l.listenconfig.Listen(ctx, "tcp", url.Host)
 	if err != nil {
 		return nil, err
 	}

src/core/version.go

@@ -8,7 +8,6 @@ import (
 	"bytes"
 	"crypto/ed25519"
 	"encoding/binary"
-	"fmt"
 	"io"
 
 	"golang.org/x/crypto/blake2b"
@@ -38,6 +37,16 @@ const (
 	metaPriority                   // uint8
 )
 
+type handshakeError string
+
+func (e handshakeError) Error() string { return string(e) }
+
+const ErrHandshakeInvalidPreamble = handshakeError("invalid handshake, remote side is not Yggdrasil")
+const ErrHandshakeInvalidLength = handshakeError("invalid handshake length, possible version mismatch")
+const ErrHandshakeInvalidPassword = handshakeError("invalid password supplied, check your config")
+const ErrHandshakeHashFailure = handshakeError("invalid hash length")
+const ErrHandshakeIncorrectPassword = handshakeError("password does not match remote side")
+
 // Gets a base metadata with no keys set, but with the correct version numbers.
 func version_getBaseMetadata() version_metadata {
 	return version_metadata{
@@ -77,7 +86,7 @@ func (m *version_metadata) encode(privateKey ed25519.PrivateKey, password []byte
 		return nil, err
 	}
 	if n != ed25519.PublicKeySize {
-		return nil, fmt.Errorf("hash writer only wrote %d bytes", n)
+		return nil, ErrHandshakeHashFailure
 	}
 	hash := hasher.Sum(nil)
 	bs = append(bs, ed25519.Sign(privateKey, hash)...)
@@ -94,11 +103,11 @@ func (m *version_metadata) decode(r io.Reader, password []byte) error {
 	}
 	meta := [4]byte{'m', 'e', 't', 'a'}
 	if !bytes.Equal(bh[:4], meta[:]) {
-		return fmt.Errorf("invalid handshake preamble")
+		return ErrHandshakeInvalidPreamble
 	}
 	hl := binary.BigEndian.Uint16(bh[4:6])
 	if hl < ed25519.SignatureSize {
-		return fmt.Errorf("invalid handshake length")
+		return ErrHandshakeInvalidLength
 	}
 	bs := make([]byte, hl)
 	if _, err := io.ReadFull(r, bs); err != nil {
@@ -132,15 +141,15 @@ func (m *version_metadata) decode(r io.Reader, password []byte) error {
 
 	hasher, err := blake2b.New512(password)
 	if err != nil {
-		return fmt.Errorf("invalid password supplied")
+		return ErrHandshakeInvalidPassword
 	}
 	n, err := hasher.Write(m.publicKey)
 	if err != nil || n != ed25519.PublicKeySize {
-		return fmt.Errorf("failed to generate hash")
+		return ErrHandshakeHashFailure
 	}
 	hash := hasher.Sum(nil)
 	if !ed25519.Verify(m.publicKey, hash, sig) {
-		return fmt.Errorf("password is incorrect")
+		return ErrHandshakeIncorrectPassword
 	}
 	return nil
 }

src/multicast/multicast.go

@@ -9,6 +9,7 @@ import (
 	"math/rand"
 	"net"
 	"net/url"
+	"sync/atomic"
 	"time"
 
 	"github.com/Arceliar/phony"
@@ -28,7 +29,7 @@ type Multicast struct {
 	core        *core.Core
 	log         *log.Logger
 	sock        *ipv6.PacketConn
-	_isOpen     bool
+	running     atomic.Bool
 	_listeners  map[string]*listenerInfo
 	_interfaces map[string]*interfaceInfo
 	_timer      *time.Timer
@@ -79,7 +80,7 @@ func New(core *core.Core, log *log.Logger, opts ...SetupOption) (*Multicast, err
 }
 
 func (m *Multicast) _start() error {
-	if m._isOpen {
+	if !m.running.CompareAndSwap(false, true) {
 		return fmt.Errorf("multicast module is already started")
 	}
 	var anyEnabled bool
@@ -87,12 +88,14 @@ func (m *Multicast) _start() error {
 		anyEnabled = anyEnabled || intf.Beacon || intf.Listen
 	}
 	if !anyEnabled {
+		m.running.Store(false)
 		return nil
 	}
 	m.log.Debugln("Starting multicast module")
 	defer m.log.Debugln("Started multicast module")
 	addr, err := net.ResolveUDPAddr("udp", string(m.config._groupAddr))
 	if err != nil {
+		m.running.Store(false)
 		return err
 	}
 	listenString := fmt.Sprintf("[::]:%v", addr.Port)
@@ -101,6 +104,7 @@ func (m *Multicast) _start() error {
 	}
 	conn, err := lc.ListenPacket(context.Background(), "udp6", listenString)
 	if err != nil {
+		m.running.Store(false)
 		return err
 	}
 	m.sock = ipv6.NewPacketConn(conn)
@@ -108,7 +112,6 @@ func (m *Multicast) _start() error {
 		// Windows can't set this flag, so we need to handle it in other ways
 	}
 
-	m._isOpen = true
 	go m.listen()
 	m.Act(nil, m._multicastStarted)
 	m.Act(nil, m._announce)
@@ -118,11 +121,7 @@ func (m *Multicast) _start() error {
 
 // IsStarted returns true if the module has been started.
 func (m *Multicast) IsStarted() bool {
-	var isOpen bool
-	phony.Block(m, func() {
-		isOpen = m._isOpen
-	})
-	return isOpen
+	return m.running.Load()
 }
 
 // Stop stops the multicast module.
@@ -136,8 +135,10 @@ func (m *Multicast) Stop() error {
 }
 
 func (m *Multicast) _stop() error {
+	if !m.running.CompareAndSwap(true, false) {
+		return nil
+	}
 	m.log.Infoln("Stopping multicast module")
-	m._isOpen = false
 	if m.sock != nil {
 		m.sock.Close()
 	}
@@ -233,7 +234,7 @@ func (m *Multicast) AnnounceNow() {
 }
 
 func (m *Multicast) _announce() {
-	if !m._isOpen {
+	if !m.running.Load() {
 		return
 	}
 	m._updateInterfaces()
@@ -250,7 +251,7 @@ func (m *Multicast) _announce() {
 	for name, info := range m._listeners {
 		// Prepare our stop function!
 		stop := func() {
-			info.listener.Close()
+			info.listener.Cancel()
 			delete(m._listeners, name)
 			m.log.Debugln("No longer multicasting on", name)
 		}
@@ -376,6 +377,9 @@ func (m *Multicast) listen() {
 	bs := make([]byte, 2048)
 	hb := make([]byte, 0, blake2b.Size) // Reused to reduce hash allocations
 	for {
+		if !m.running.Load() {
+			return
+		}
 		n, rcm, fromAddr, err := m.sock.ReadFrom(bs)
 		if err != nil {
 			if !m.IsStarted() {

src/multicast/multicast_darwin_cgo.go

@@ -31,7 +31,7 @@ import (
 )
 
 func (m *Multicast) _multicastStarted() {
-	if !m._isOpen {
+	if !m.running.Load() {
 		return
 	}
 	C.StopAWDLBrowsing()

src/tun/tun.go

@@ -11,7 +11,6 @@ import (
 	"io"
 	"net"
 	"sync"
-	"time"
 
 	"github.com/Arceliar/phony"
 	wgtun "golang.zx2c4.com/wireguard/tun"
@@ -65,20 +64,6 @@ func getSupportedMTU(mtu uint64) uint64 {
 	return mtu
 }
 
-func waitForTUNUp(ch <-chan wgtun.Event) bool {
-	t := time.After(time.Second * 5)
-	for {
-		select {
-		case ev := <-ch:
-			if ev == wgtun.EventUp {
-				return true
-			}
-		case <-t:
-			return false
-		}
-	}
-}
-
 // Name returns the name of the adapter, e.g. "tun0". On Windows, this may
 // return a canonical adapter name instead.
 func (tun *TunAdapter) Name() string {

src/tun/tun_bsd.go

@@ -80,9 +80,6 @@ func (tun *TunAdapter) setup(ifname string, addr string, mtu uint64) error {
 	if err != nil {
 		return fmt.Errorf("failed to create TUN: %w", err)
 	}
-	if !waitForTUNUp(iface.Events()) {
-		return fmt.Errorf("TUN did not come up in time")
-	}
 	tun.iface = iface
 	if mtu, err := iface.MTU(); err == nil {
 		tun.mtu = getSupportedMTU(uint64(mtu))

src/tun/tun_darwin.go

@@ -27,9 +27,6 @@ func (tun *TunAdapter) setup(ifname string, addr string, mtu uint64) error {
 	if err != nil {
 		return fmt.Errorf("failed to create TUN: %w", err)
 	}
-	if !waitForTUNUp(iface.Events()) {
-		return fmt.Errorf("TUN did not come up in time")
-	}
 	tun.iface = iface
 	if m, err := iface.MTU(); err == nil {
 		tun.mtu = getSupportedMTU(uint64(m))
@@ -58,9 +55,6 @@ func (tun *TunAdapter) setupFD(fd int32, addr string, mtu uint64) error {
 		unix.Close(dfd)
 		return fmt.Errorf("failed to create TUN from FD: %w", err)
 	}
-	if !waitForTUNUp(iface.Events()) {
-		return fmt.Errorf("TUN did not come up in time")
-	}
 	tun.iface = iface
 	if m, err := iface.MTU(); err == nil {
 		tun.mtu = getSupportedMTU(uint64(m))

src/tun/tun_linux.go

@@ -21,9 +21,6 @@ func (tun *TunAdapter) setup(ifname string, addr string, mtu uint64) error {
 	if err != nil {
 		return fmt.Errorf("failed to create TUN: %w", err)
 	}
-	if !waitForTUNUp(iface.Events()) {
-		return fmt.Errorf("TUN did not come up in time")
-	}
 	tun.iface = iface
 	if mtu, err := iface.MTU(); err == nil {
 		tun.mtu = getSupportedMTU(uint64(mtu))

src/tun/tun_other.go

@@ -18,9 +18,6 @@ func (tun *TunAdapter) setup(ifname string, addr string, mtu uint64) error {
 	if err != nil {
 		return fmt.Errorf("failed to create TUN: %w", err)
 	}
-	if !waitForTUNUp(iface.Events()) {
-		return fmt.Errorf("TUN did not come up in time")
-	}
 	tun.iface = iface
 	if mtu, err := iface.MTU(); err == nil {
 		tun.mtu = getSupportedMTU(uint64(mtu))

src/tun/tun_windows.go

@@ -8,10 +8,12 @@ import (
 	"fmt"
 	"log"
 	"net/netip"
+	"time"
 
 	"github.com/yggdrasil-network/yggdrasil-go/src/config"
 	"golang.org/x/sys/windows"
 
+	"golang.zx2c4.com/wintun"
 	wgtun "golang.zx2c4.com/wireguard/tun"
 	"golang.zx2c4.com/wireguard/windows/elevate"
 	"golang.zx2c4.com/wireguard/windows/tunnel/winipcfg"
@@ -31,14 +33,23 @@ func (tun *TunAdapter) setup(ifname string, addr string, mtu uint64) error {
 		if guid, err = windows.GUIDFromString("{8f59971a-7872-4aa6-b2eb-061fc4e9d0a7}"); err != nil {
 			return err
 		}
-		if iface, err = wgtun.CreateTUNWithRequestedGUID(ifname, &guid, int(mtu)); err != nil {
-			return err
-		}
-		if !waitForTUNUp(iface.Events()) {
-			return fmt.Errorf("TUN did not come up in time")
+		iface, err = wgtun.CreateTUNWithRequestedGUID(ifname, &guid, int(mtu))
+		if err != nil {
+			// Very rare condition, it will purge the old device and create new
+			tun.log.Printf("Error creating TUN: '%s'", err)
+			wintun.Uninstall()
+			time.Sleep(3 * time.Second)
+			tun.log.Printf("Trying again")
+			iface, err = wgtun.CreateTUNWithRequestedGUID(ifname, &guid, int(mtu))
+			if err != nil {
+				return err
+			}
 		}
+		tun.log.Printf("Waiting for TUN to come up")
+		time.Sleep(1 * time.Second)
 		tun.iface = iface
 		if addr != "" {
+			tun.log.Printf("Setting up address")
 			if err = tun.setupAddress(addr); err != nil {
 				tun.log.Errorln("Failed to set up TUN address:", err)
 				return err
@@ -51,6 +62,7 @@ func (tun *TunAdapter) setup(ifname string, addr string, mtu uint64) error {
 		if mtu, err := iface.MTU(); err == nil {
 			tun.mtu = uint64(mtu)
 		}
+		tun.log.Printf("TUN is set up successfully")
 		return nil
 	})
 }