Merge pull request #176 from fosrl/dev

1.10.2

main.go

@@ -238,7 +238,8 @@ func runOlmMainWithArgs(ctx context.Context, cancel context.CancelFunc, signalCt
 		SocketPath:   config.SocketPath,
 		Version:      config.Version,
 		Agent:        "Olm CLI",
-		OnExit:       cancel, // Pass cancel function directly to trigger shutdown
+		OlmVersion:   config.Version, // the standalone binary is olm itself
+		OnExit:       cancel,         // Pass cancel function directly to trigger shutdown
 		OnTerminated: cancel,
 		PprofAddr:    ":4444", // TODO: REMOVE OR MAKE CONFIGURABLE
 		// Re-invoke this binary in watchdog mode to clean up DNS if we die.

olm/olm.go

@@ -288,6 +288,8 @@ func Init(ctx context.Context, config OlmConfig) (*Olm, error) {
 	apiServer.SetVersion(config.Version)
 	apiServer.SetAgent(config.Agent)
 
+	config.OlmVersion = resolveOlmVersion(config)
+
 	newOlm := &Olm{
 		logFile:                logFile,
 		olmCtx:                 ctx,
@@ -727,16 +729,17 @@ func (o *Olm) StartTunnel(config TunnelConfig) {
 		publicKey := o.privateKey.PublicKey()
 		logger.Debug("Sending registration message to server with public key: %s, relay: %v, pingResults: %+v", publicKey, !config.Holepunch, pingResults)
 		o.stopRegister, o.updateRegister = o.websocket.SendMessageInterval("olm/wg/register", map[string]any{
-			"publicKey":   publicKey.String(),
+			"publicKey":    publicKey.String(),
-			"relay":       !config.Holepunch,
+			"relay":        !config.Holepunch,
-			"olmVersion":  o.olmConfig.Version,
+			"olmVersion":   o.olmConfig.OlmVersion,
-			"olmAgent":    o.olmConfig.Agent,
+			"olmAgent":     o.olmConfig.Agent,
-			"orgId":       config.OrgID,
+			"agentVersion": o.olmConfig.Version,
-			"userToken":   userToken,
+			"orgId":        config.OrgID,
-			"fingerprint": o.fingerprint,
+			"userToken":    userToken,
-			"postures":    o.postures,
+			"fingerprint":  o.fingerprint,
-			"pingResults": pingResults,
+			"postures":     o.postures,
-			"chainId":     generateChainId(), // use a random chainId for registration updates - it won't be used for cancellation since registration is a one-time message but for tracking the session
+			"pingResults":  pingResults,
+			"chainId":      generateChainId(), // use a random chainId for registration updates - it won't be used for cancellation since registration is a one-time message but for tracking the session
 		}, 2*time.Second, 20) // after 18 tries on the server side we send the error so dont change this without changing that
 	})
 
@@ -786,8 +789,9 @@ func (o *Olm) StartTunnel(config TunnelConfig) {
 			if err := o.websocket.SendMessage("olm/wg/register", map[string]any{
 				"publicKey":           publicKey.String(),
 				"relay":               !config.Holepunch,
-				"olmVersion":          o.olmConfig.Version,
+				"olmVersion":          o.olmConfig.OlmVersion,
 				"olmAgent":            o.olmConfig.Agent,
+				"agentVersion":        o.olmConfig.Version,
 				"orgId":               config.OrgID,
 				"userToken":           userToken,
 				"fingerprint":         o.fingerprint,

olm/types.go

@@ -82,8 +82,11 @@ type OlmConfig struct {
 	EnableAPI  bool
 	HTTPAddr   string
 	SocketPath string
-	Version    string
+	Version    string // version of the agent (the app embedding olm)
 	Agent      string
+	// OlmVersion is the version of the github.com/fosrl/olm module itself.
+	// Optional: when empty it is read from the binary's build info.
+	OlmVersion string
 
 	WakeUpDebounce time.Duration
 

olm/version.go

@@ -0,0 +1,30 @@
+package olm
+
+import (
+	"runtime/debug"
+	"strings"
+)
+
+// olmModulePath is the module path of this package, used to find its version
+// in the build info of a binary that embeds olm (the CLI, Pangolin Windows, ...).
+const olmModulePath = "github.com/fosrl/olm"
+
+// resolveOlmVersion returns the version of the olm module this binary was
+// built with, as opposed to config.Version which is the embedding agent's
+// version. It prefers config.OlmVersion, then the module version from the
+// binary's build info, and returns "" if neither is available.
+func resolveOlmVersion(config OlmConfig) string {
+	if config.OlmVersion != "" {
+		return config.OlmVersion
+	}
+
+	if bi, ok := debug.ReadBuildInfo(); ok {
+		for _, dep := range bi.Deps {
+			if dep.Path == olmModulePath {
+				return strings.TrimPrefix(dep.Version, "v")
+			}
+		}
+	}
+
+	return ""
+}

peers/monitor/monitor.go

@@ -481,7 +481,7 @@ func (pm *PeerMonitor) rapidTestOnLocalFallback(siteID int, publicEndpoint strin
 
 	pm.mutex.Lock()
 	_, stillTracked := pm.localEndpoints[siteID]
-	noLocalActiveYet := pm.localActiveEndpoint[siteID] == ""
+	winnerStillActive := pm.localActiveEndpoint[siteID] == localWinner
 	switchCb := pm.localSwitchCallback
 	pm.mutex.Unlock()
 
@@ -492,8 +492,8 @@ func (pm *PeerMonitor) rapidTestOnLocalFallback(siteID int, publicEndpoint strin
 	if localWinner != "" {
 		// RapidTestLocalEndpoints already recorded the new active endpoint and notified the
 		// server, but doesn't move the WireGuard peer itself - do that here, unless a
-		// concurrent checkLocalEndpoints tick already beat us to activating something.
+		// concurrent checkLocalEndpoints tick has since activated a different endpoint.
-		if noLocalActiveYet && switchCb != nil {
+		if winnerStillActive && switchCb != nil {
 			switchCb(siteID, localWinner)
 		}
 		logger.Info("Rapid fallback test: local connection %s viable for site %d", localWinner, siteID)

peers/monitor/monitor_test.go

@@ -0,0 +1,129 @@
+package monitor
+
+import (
+	"bytes"
+	"net"
+	"testing"
+	"time"
+
+	"github.com/fosrl/newt/bind"
+	wgConn "golang.zx2c4.com/wireguard/conn"
+)
+
+// listenLoopback opens a UDP socket on 127.0.0.1 that is closed when the test ends.
+func listenLoopback(t *testing.T) *net.UDPConn {
+	t.Helper()
+	conn, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 0})
+	if err != nil {
+		t.Fatalf("failed to listen on loopback: %v", err)
+	}
+	t.Cleanup(func() { conn.Close() })
+	return conn
+}
+
+// startReachableEndpoint returns the address of a loopback endpoint that answers holepunch
+// test requests the way a site does.
+func startReachableEndpoint(t *testing.T) string {
+	t.Helper()
+	conn := listenLoopback(t)
+	go func() {
+		buf := make([]byte, 1500)
+		for {
+			n, addr, err := conn.ReadFromUDP(buf)
+			if err != nil {
+				return
+			}
+			if n < bind.MagicTestRequestLen || !bytes.HasPrefix(buf[:n], bind.MagicTestRequest) {
+				continue
+			}
+			resp := make([]byte, bind.MagicTestResponseLen)
+			copy(resp, bind.MagicTestResponse)
+			copy(resp[len(bind.MagicTestResponse):], buf[len(bind.MagicTestRequest):bind.MagicTestRequestLen])
+			_, _ = conn.WriteToUDP(resp, addr)
+		}
+	}()
+	return conn.LocalAddr().String()
+}
+
+// startUnreachableEndpoint returns the address of a loopback endpoint that never answers,
+// standing in for an endpoint that can't be reached from this client.
+func startUnreachableEndpoint(t *testing.T) string {
+	t.Helper()
+	return listenLoopback(t).LocalAddr().String()
+}
+
+// newTestPeerMonitor returns a PeerMonitor whose holepunch tester sends its probes from a
+// loopback SharedBind. Only the tester is started, not the monitor loops, so the test drives
+// the monitor by hand.
+func newTestPeerMonitor(t *testing.T) *PeerMonitor {
+	t.Helper()
+
+	sharedBind, err := bind.New(listenLoopback(t))
+	if err != nil {
+		t.Fatalf("failed to create SharedBind: %v", err)
+	}
+	t.Cleanup(func() { sharedBind.Close() })
+
+	// SharedBind only delivers test responses to the tester from its receive path, which
+	// WireGuard would normally be calling.
+	recvFuncs, _, err := sharedBind.Open(0)
+	if err != nil {
+		t.Fatalf("failed to open SharedBind: %v", err)
+	}
+	go func() {
+		bufs := [][]byte{make([]byte, 1500)}
+		sizes := make([]int, 1)
+		eps := make([]wgConn.Endpoint, 1)
+		for {
+			if _, err := recvFuncs[0](bufs, sizes, eps); err != nil {
+				return
+			}
+		}
+	}()
+
+	pm := NewPeerMonitor(nil, nil, "", sharedBind, nil, nil)
+	pm.rapidTestTimeout = 500 * time.Millisecond
+	pm.rapidTestInterval = 10 * time.Millisecond
+	pm.rapidTestMaxAttempts = 1
+
+	if err := pm.holepunchTester.Start(); err != nil {
+		t.Fatalf("failed to start holepunch tester: %v", err)
+	}
+	t.Cleanup(pm.holepunchTester.Stop)
+
+	return pm
+}
+
+// A site's active local endpoint stopped answering, the public endpoint is unreachable too
+// (e.g. the client is on the site's LAN and the router doesn't hairpin), but another local
+// candidate answers. The WireGuard peer has to be moved to that candidate.
+func TestRapidTestOnLocalFallbackSwitchesToLocalWinner(t *testing.T) {
+	const siteID = 1
+
+	pm := newTestPeerMonitor(t)
+
+	deadLocal := startUnreachableEndpoint(t)
+	liveLocal := startReachableEndpoint(t)
+	public := startUnreachableEndpoint(t)
+
+	localEndpoints := []string{deadLocal, liveLocal}
+	// The monitor address is inside the tunnel; it isn't dialed since the monitor isn't started.
+	if err := pm.AddPeer(siteID, "100.90.128.1:21821", public, localEndpoints); err != nil {
+		t.Fatalf("AddPeer failed: %v", err)
+	}
+
+	var switchedTo []string
+	pm.SetLocalConnectionCallbacks(
+		func(_ int, endpoint string) { switchedTo = append(switchedTo, endpoint) },
+		func(int) {},
+	)
+
+	// State right after checkLocalEndpoints gave up on deadLocal: no local endpoint active,
+	// the WireGuard peer back on the public endpoint.
+	pm.rapidTestOnLocalFallback(siteID, public, remainingLocalCandidates(localEndpoints, deadLocal))
+
+	if len(switchedTo) != 1 || switchedTo[0] != liveLocal {
+		t.Fatalf("local switch callback calls = %q, want [%q] (monitor's active local endpoint: %q)",
+			switchedTo, liveLocal, pm.localActiveEndpoint[siteID])
+	}
+}