From 2f6c988125b26500c36dc476fd2139a12d6d368b Mon Sep 17 00:00:00 2001 From: Pavel Sviderski Date: Sun, 13 Oct 2024 01:15:32 -0400 Subject: [PATCH] do not rotate endpoints on daemon restart, handle endpoint changes on established reverse peer connections --- internal/machine/network/peer.go | 29 +++- internal/machine/network/wireguard_linux.go | 170 ++++++++++++-------- 2 files changed, 133 insertions(+), 66 deletions(-) diff --git a/internal/machine/network/peer.go b/internal/machine/network/peer.go index 3a20888a..1c7d8c2e 100644 --- a/internal/machine/network/peer.go +++ b/internal/machine/network/peer.go @@ -23,13 +23,21 @@ type peer struct { status string } -func newPeer(config PeerConfig) *peer { +func newPeer(config PeerConfig, wgPeer *wgtypes.Peer) *peer { p := &peer{ config: config, status: PeerStatusUnknown, } if p.config.Endpoint != nil { p.lastEndpointChangeTime = time.Now() + // Reset the endpoint change time if the endpoint is the same as the one in the current WireGuard peer. + // This is to avoid unnecessary endpoint rotation for the already connected peer. + if wgPeer != nil && wgPeer.Endpoint != nil { + wgEndpoint := wgPeer.Endpoint.AddrPort() + if *p.config.Endpoint == wgEndpoint { + p.lastEndpointChangeTime = time.Time{} + } + } } return p } @@ -42,11 +50,25 @@ func (p *peer) updateConfig(config PeerConfig) { p.config = config } -func (p *peer) updateFromWireGuard(wgPeer wgtypes.Peer) { +func (p *peer) updateFromDevice(wgPeer wgtypes.Peer) (endpointChanged bool) { + if wgPeer.Endpoint != nil { + wgEndpoint := wgPeer.Endpoint.AddrPort() + if p.config.Endpoint == nil || *p.config.Endpoint != wgEndpoint { + // The peer endpoint has been automatically updated on the WireGuard device which normally happens + // when the peer establishes a reverse connection to this machine. + p.config.Endpoint = &wgEndpoint + // Reset the endpoint change time to not attempt to rotate the endpoint and correctly calculate the status. + p.lastEndpointChangeTime = time.Time{} + endpointChanged = true + slog.Info("Peer endpoint automatically updated on WireGuard interface by establishing a reverse "+ + "connection to this machine.", "public_key", p.config.PublicKey, "endpoint", wgEndpoint) + } + } p.lastHandshakeTime = wgPeer.LastHandshakeTime p.receiveBytes = wgPeer.ReceiveBytes p.transmitBytes = wgPeer.TransmitBytes p.calculateStatus() + return } // Peer status calculation is based on Talos Kubespan implementation: @@ -119,7 +141,8 @@ func (p *peer) calculateStatus() { p.status = PeerStatusUnknown } if p.status != lastStatus { - slog.Info("Peer status changed.", "public_key", p.config.PublicKey, "status", p.status) + slog.Info("Peer status changed.", "public_key", p.config.PublicKey, + "status", p.status, "previous_status", lastStatus) } } diff --git a/internal/machine/network/wireguard_linux.go b/internal/machine/network/wireguard_linux.go index b3c21f94..dba6b309 100644 --- a/internal/machine/network/wireguard_linux.go +++ b/internal/machine/network/wireguard_linux.go @@ -35,10 +35,7 @@ func NewWireGuardNetwork() (*WireGuardNetwork, error) { if err != nil { return nil, fmt.Errorf("create or get WireGuard link %q: %v", WireGuardInterfaceName, err) } - return &WireGuardNetwork{ - link: link, - peers: make(map[string]*peer), - }, nil + return &WireGuardNetwork{link: link}, nil } // createOrGetLink creates a new WireGuard link with the given name if it doesn't already exist, otherwise it returns the existing link. @@ -76,48 +73,11 @@ func (n *WireGuardNetwork) Configure(config Config) error { n.mu.Lock() defer n.mu.Unlock() - // Create or update peer structs based on the config. - newPeersSet := make(map[string]struct{}, len(config.Peers)) - for _, pc := range config.Peers { - if p, ok := n.peers[pc.PublicKey.String()]; ok { - p.updateConfig(pc) - } else { - n.peers[pc.PublicKey.String()] = newPeer(pc) - } - newPeersSet[pc.PublicKey.String()] = struct{}{} - } - // Delete peers that are no longer in the config. - for k := range n.peers { - if _, ok := newPeersSet[k]; !ok { - delete(n.peers, k) - } - } - - wg, err := wgctrl.New() - if err != nil { - return fmt.Errorf("create WireGuard client: %w", err) - } - defer wg.Close() - - dev, err := wg.Device(n.link.Attrs().Name) - if err != nil { - return fmt.Errorf("get WireGuard device %q: %w", n.link.Attrs().Name, err) - } - wgConfig, err := config.toDeviceConfig(dev.Peers) - if err != nil { + if err := n.configureDevice(config); err != nil { return err } - // Apply the new configuration to the WireGuard device. - if err = wg.ConfigureDevice(n.link.Attrs().Name, wgConfig); err != nil { - return fmt.Errorf("configure WireGuard device %q: %w", n.link.Attrs().Name, err) - } slog.Info("Configured WireGuard interface.", "name", n.link.Attrs().Name) - if err = n.updatePeersFromWireGuard(); err != nil { - return err - } - slog.Debug("Updated peers status from WireGuard interface.", "name", n.link.Attrs().Name) - machinePrefix := netip.PrefixFrom(MachineIP(config.Subnet), config.Subnet.Bits()) managementPrefix, err := addrToSingleIPPrefix(config.ManagementIP) if err != nil { @@ -150,29 +110,62 @@ func (n *WireGuardNetwork) Configure(config Config) error { return nil } -// updatePeersFromWireGuard updates the peers status from the WireGuard device peers. -// mu lock must be held before calling this method. -func (n *WireGuardNetwork) updatePeersFromWireGuard() error { +func (n *WireGuardNetwork) configureDevice(config Config) error { wg, err := wgctrl.New() if err != nil { return fmt.Errorf("create WireGuard client: %w", err) } defer wg.Close() + // Get the current WireGuard peers from the device which are required to reconstruct the local peers structs + // and build a new config for the device. dev, err := wg.Device(n.link.Attrs().Name) if err != nil { return fmt.Errorf("get WireGuard device %q: %w", n.link.Attrs().Name, err) } - - for _, wgPeer := range dev.Peers { - publicKey := secret.Secret(wgPeer.PublicKey[:]) - if p, ok := n.peers[publicKey.String()]; ok { - p.updateFromWireGuard(wgPeer) - } else { - // Assume that WG peers are not updated out of band so they should always be in sync with the config. - slog.Warn("Found WireGuard peer that is not in the configuration.", "public_key", publicKey) + if n.peers == nil { + // This is the first time we configure this instance of the network. If the device has been configured earlier + // and the daemon restarted, attempt to reconstruct the peers from the device to avoid unnecessary endpoint + // rotations and connection disruptions. + n.peers = make(map[string]*peer, len(config.Peers)) + wgPeers := make(map[string]*wgtypes.Peer, len(dev.Peers)) + for i, _ := range dev.Peers { + wgPeers[secret.Secret(dev.Peers[i].PublicKey[:]).String()] = &dev.Peers[i] + } + for _, pc := range config.Peers { + wgPeer := wgPeers[pc.PublicKey.String()] + n.peers[pc.PublicKey.String()] = newPeer(pc, wgPeer) } } + + // Create or update peer structs based on the config. + newPeersSet := make(map[string]struct{}, len(config.Peers)) + for _, pc := range config.Peers { + if p, ok := n.peers[pc.PublicKey.String()]; ok { + p.updateConfig(pc) + } else { + // WireGuard peers should not be configured out of band so the current WG peer is provided only + // for the first time configuration above. + n.peers[pc.PublicKey.String()] = newPeer(pc, nil) + } + newPeersSet[pc.PublicKey.String()] = struct{}{} + } + // Delete peers that are no longer in the config. + for k := range n.peers { + if _, ok := newPeersSet[k]; !ok { + delete(n.peers, k) + } + } + + wgConfig, err := config.toDeviceConfig(dev.Peers) + if err != nil { + return err + } + // Apply the new configuration to the WireGuard device. + if err = wg.ConfigureDevice(n.link.Attrs().Name, wgConfig); err != nil { + return fmt.Errorf("configure WireGuard device %q: %w", n.link.Attrs().Name, err) + } + return nil } @@ -284,12 +277,12 @@ func (n *WireGuardNetwork) Run(ctx context.Context) error { select { case <-ticker.C: n.mu.Lock() - if err = n.changeWireGuardEndpoints(ctx); err != nil { - slog.Error("Failed to update peer endpoints on WireGuard interface.", + if err = n.updatePeersFromDevice(ctx); err != nil { + slog.Error("Failed to update peers status from WireGuard interface.", "name", n.link.Attrs().Name, "err", err) } - if err = n.updatePeersFromWireGuard(); err != nil { - slog.Error("Failed to update peers status from WireGuard interface.", + if err = n.changeWireGuardEndpoints(ctx); err != nil { + slog.Error("Failed to update peer endpoints on WireGuard interface.", "name", n.link.Attrs().Name, "err", err) } n.mu.Unlock() @@ -309,8 +302,50 @@ func (n *WireGuardNetwork) WatchEndpoints() <-chan EndpointChangeEvent { return ch } +// updatePeersFromDevice updates the peers status from the WireGuard device peers. +// mu lock must be held before calling this method. +func (n *WireGuardNetwork) updatePeersFromDevice(ctx context.Context) error { + wg, err := wgctrl.New() + if err != nil { + return fmt.Errorf("create WireGuard client: %w", err) + } + defer wg.Close() + + dev, err := wg.Device(n.link.Attrs().Name) + if err != nil { + return fmt.Errorf("get WireGuard device %q: %w", n.link.Attrs().Name, err) + } + + var events []EndpointChangeEvent + for _, wgPeer := range dev.Peers { + publicKey := secret.Secret(wgPeer.PublicKey[:]) + if p, ok := n.peers[publicKey.String()]; ok { + endpointChanged := p.updateFromDevice(wgPeer) + if endpointChanged { + events = append(events, EndpointChangeEvent{ + PublicKey: publicKey, + Endpoint: *p.config.Endpoint, + }) + } + } else { + // Assume that WG peers are not updated out of band so they should always be in sync with the config. + slog.Warn("Found WireGuard peer that is not in the configuration.", "public_key", publicKey) + } + } + + if len(events) > 0 { + if err = n.notifyWatchers(ctx, events); err != nil { + // As of October 2024, the machine is the only watcher to persist changed endpoints in the machine + // state which can tolerate missed events. Therefore, we can ignore the error here. + slog.Error("Failed to notify watchers about a peer endpoint change.", "err", err) + } + } + return nil +} + // changeWireGuardEndpoints rotates the endpoints of the WireGuard peers with 'down' status // in an attempt to find a working one. +// mu lock must be held before calling this method. func (n *WireGuardNetwork) changeWireGuardEndpoints(ctx context.Context) error { var wgPeerConfigs []wgtypes.PeerConfig var events []EndpointChangeEvent @@ -361,21 +396,30 @@ func (n *WireGuardNetwork) changeWireGuardEndpoints(ctx context.Context) error { return fmt.Errorf("configure WireGuard device %q with endpoint changes: %w", n.link.Attrs().Name, err) } for _, pc := range wgPeerConfigs { + publicKey := secret.Secret(pc.PublicKey[:]) slog.Info("Changed peer endpoint on WireGuard interface.", - "name", n.link.Attrs().Name, "public_key", secret.Secret(pc.PublicKey[:]), "endpoint", pc.Endpoint) + "name", n.link.Attrs().Name, "public_key", publicKey, "endpoint", pc.Endpoint, + "status", n.peers[publicKey.String()].status) } - // Notify the watchers about the endpoint changes. + if err = n.notifyWatchers(ctx, events); err != nil { + // As of October 2024, the machine is the only watcher to persist changed endpoints in the machine + // state which can tolerate missed events. Therefore, we can ignore the error here. + slog.Error("Failed to notify watchers about a peer endpoint change.", "err", err) + } + + return nil +} + +// notifyWatchers notifies the watchers about peer endpoint changes. +func (n *WireGuardNetwork) notifyWatchers(ctx context.Context, events []EndpointChangeEvent) error { for _, ch := range n.watchers { for _, e := range events { select { case ch <- e: // Use a timeout to avoid blocking the network control loop. case <-time.After(1 * time.Second): - slog.Error("Timeout notifying watchers about a peer endpoint change.") - // As of October 2024, the machine is the only watcher to persist changed endpoints in the machine - // state which can tolerate missed events. Therefore, we can ignore the error here. - return nil + return errors.New("timeout 1 second") case <-ctx.Done(): return nil }