From e074b6855fb0ddecbaa6daf7e17f610de23dbd89 Mon Sep 17 00:00:00 2001 From: Pavel Sviderski Date: Mon, 26 Aug 2024 17:09:54 +1000 Subject: [PATCH] implement uncloudd daemon that starts a WireGuard network without peer probing --- cmd/machined/install/root.go | 41 ---- cmd/machined/main.go | 26 --- cmd/uncloudd/main.go | 43 ++++ internal/cli/cluster.go | 7 + internal/machine/daemon/config.go | 45 ----- internal/machine/daemon/daemon.go | 34 ++++ internal/machine/daemon/run.go | 17 -- internal/machine/install.go | 15 -- internal/machine/network/config.go | 2 +- internal/machine/network/ip.go | 18 ++ internal/machine/network/wireguard.go | 27 +++ internal/machine/network/wireguard_darwin.go | 22 +++ internal/machine/network/wireguard_linux.go | 195 +++++++++++++++++++ 13 files changed, 347 insertions(+), 145 deletions(-) delete mode 100644 cmd/machined/install/root.go delete mode 100644 cmd/machined/main.go create mode 100644 cmd/uncloudd/main.go delete mode 100644 internal/machine/daemon/config.go create mode 100644 internal/machine/daemon/daemon.go delete mode 100644 internal/machine/daemon/run.go delete mode 100644 internal/machine/install.go create mode 100644 internal/machine/network/ip.go create mode 100644 internal/machine/network/wireguard.go create mode 100644 internal/machine/network/wireguard_darwin.go create mode 100644 internal/machine/network/wireguard_linux.go diff --git a/cmd/machined/install/root.go b/cmd/machined/install/root.go deleted file mode 100644 index 90208b3d..00000000 --- a/cmd/machined/install/root.go +++ /dev/null @@ -1,41 +0,0 @@ -package install - -import ( - "github.com/spf13/cobra" - "uncloud/internal/machine" - "uncloud/internal/machine/daemon" -) - -type Options struct { - uncloudID string - uncloudSecret string - network string -} - -func NewCommand(dataDir *string) *cobra.Command { - opts := Options{} - cmd := &cobra.Command{ - Use: "install", - Short: "Install OS dependencies and configure Uncloud machine.", - RunE: func(cmd *cobra.Command, args []string) error { - return install(*dataDir, opts) - }, - } - cmd.Flags().StringVar(&opts.uncloudID, "id", "", "Globally unique identifier for the uncloud this machine belongs to") - _ = cmd.MarkFlagRequired("id") - // TODO: read secret from file for security reasons (bash history). - cmd.Flags().StringVar(&opts.uncloudSecret, "secret", "", "Shared secret for the uncloud this machine belongs to") - _ = cmd.MarkFlagRequired("secret") - cmd.Flags().StringVar(&opts.network, "network", "", "IPv4 network in CIDR format to use for the machine network") - _ = cmd.MarkFlagRequired("network") - return cmd -} - -func install(dataDir string, opts Options) error { - cfg := daemon.Config{ - UncloudID: opts.uncloudID, - UncloudSecret: opts.uncloudSecret, - Network: opts.network, - } - return machine.Install(dataDir, cfg) -} diff --git a/cmd/machined/main.go b/cmd/machined/main.go deleted file mode 100644 index be6b4c41..00000000 --- a/cmd/machined/main.go +++ /dev/null @@ -1,26 +0,0 @@ -package main - -import ( - "github.com/spf13/cobra" - "uncloud/cmd/machined/install" - "uncloud/internal/machine/daemon" -) - -func main() { - var dataDir string - cmd := &cobra.Command{ - Use: "machined", - Short: "Uncloud machine daemon.", - SilenceUsage: true, - SilenceErrors: true, - RunE: func(cmd *cobra.Command, args []string) error { - return daemon.Run(dataDir) - }, - } - cmd.PersistentFlags().StringVarP(&dataDir, "data-dir", "d", daemon.DefaultDataDir, "Directory to store machine state") - _ = cmd.MarkFlagDirname("data-dir") - cmd.AddCommand( - install.NewCommand(&dataDir), - ) - cobra.CheckErr(cmd.Execute()) -} diff --git a/cmd/uncloudd/main.go b/cmd/uncloudd/main.go new file mode 100644 index 00000000..05ecdb37 --- /dev/null +++ b/cmd/uncloudd/main.go @@ -0,0 +1,43 @@ +package main + +import ( + "context" + "github.com/spf13/cobra" + "log/slog" + "os" + "os/signal" + "syscall" + "uncloud/internal/machine" + "uncloud/internal/machine/daemon" +) + +func main() { + var dataDir string + cmd := &cobra.Command{ + Use: "uncloudd", + Short: "Uncloud machine daemon.", + SilenceUsage: true, + SilenceErrors: true, + RunE: func(cmd *cobra.Command, args []string) error { + return daemon.Run(cmd.Context(), dataDir) + }, + } + cmd.PersistentFlags().StringVarP(&dataDir, "data-dir", "d", machine.DefaultDataDir, + "Directory for storing persistent machine state") + _ = cmd.MarkFlagDirname("data-dir") + + // ctx is canceled when the daemon command is interrupted. + ctx, cancel := context.WithCancel(context.Background()) + + // Handle interrupt signals and cancel the context. + sigs := make(chan os.Signal, 1) + signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM) + go func() { + sig := <-sigs + slog.Info("Received signal, stopping daemon.", "signal", sig) + cancel() + }() + + cobra.CheckErr(cmd.ExecuteContext(ctx)) + slog.Info("Daemon stopped.") +} diff --git a/internal/cli/cluster.go b/internal/cli/cluster.go index 55671fed..f31e51c8 100644 --- a/internal/cli/cluster.go +++ b/internal/cli/cluster.go @@ -144,10 +144,17 @@ func (c *Cluster) AddMachine(ctx context.Context, name, user, host string, port if err != nil { return "", fmt.Errorf("write machine config to %q: %w", mcfgPath, err) } + fmt.Println("Machine config written to", mcfgPath) // TODO: download and install the latest uncloudd binary by running the install shell script from GitHub. // For now upload the binary using scp manually. + out, err := exec.Run(ctx, cmdexec.QuoteCommand(sudoPrefix, "systemctl", "restart", "uncloudd")) + if err != nil { + return "", fmt.Errorf("start uncloudd: %w: %s", err, out) + } + fmt.Println("uncloudd started") + connConfig := config.MachineConnection{ User: user, Host: host, diff --git a/internal/machine/daemon/config.go b/internal/machine/daemon/config.go deleted file mode 100644 index 9a740f1f..00000000 --- a/internal/machine/daemon/config.go +++ /dev/null @@ -1,45 +0,0 @@ -package daemon - -import ( - "encoding/json" - "fmt" - "os" - "path/filepath" -) - -const ( - DefaultDataDir = "/var/lib/uncloud" - MachineConfigPath = "machine.json" -) - -type Config struct { - UncloudID string - UncloudSecret string - // IPv4 network in CIDR format to use for the machine network. - Network string -} - -func ReadConfig(path string) (Config, error) { - data, err := os.ReadFile(path) - if err != nil { - return Config{}, fmt.Errorf("read config file %q: %w", path, err) - } - var config Config - if err = json.Unmarshal(data, &config); err != nil { - return Config{}, fmt.Errorf("parse config file %q: %w", path, err) - } - return config, nil -} - -func (c *Config) Write(path string) error { - dir, _ := filepath.Split(path) - if err := os.MkdirAll(dir, 0700); err != nil { - return fmt.Errorf("create config directory %q: %w", dir, err) - } - - data, err := json.Marshal(c) - if err != nil { - return fmt.Errorf("marshal config: %w", err) - } - return os.WriteFile(path, data, 0600) -} diff --git a/internal/machine/daemon/daemon.go b/internal/machine/daemon/daemon.go new file mode 100644 index 00000000..7806f9e9 --- /dev/null +++ b/internal/machine/daemon/daemon.go @@ -0,0 +1,34 @@ +package daemon + +import ( + "context" + "fmt" + "uncloud/internal/machine" + "uncloud/internal/machine/network" +) + +func Run(ctx context.Context, dataDir string) error { + cfg, err := machine.ParseConfig(machine.ConfigPath(dataDir)) + if err != nil { + return fmt.Errorf("load machine config: %w", err) + } + + wgnet, err := network.NewWireGuardNetwork() + if err != nil { + return fmt.Errorf("create WireGuard network: %w", err) + } + if err = wgnet.Configure(*cfg.Network); err != nil { + return fmt.Errorf("configure WireGuard network: %w", err) + } + + //ctx, cancel := context.WithCancel(context.Background()) + //go wgnet.WatchEndpoints(ctx, peerEndpointChangeNotifier) + + //addrs, err := network.ListRoutableAddresses() + //if err != nil { + // return err + //} + //fmt.Println("Addresses:", addrs) + + return wgnet.Run(ctx) +} diff --git a/internal/machine/daemon/run.go b/internal/machine/daemon/run.go deleted file mode 100644 index c678132e..00000000 --- a/internal/machine/daemon/run.go +++ /dev/null @@ -1,17 +0,0 @@ -package daemon - -import ( - "errors" - "fmt" - "path/filepath" -) - -func Run(dataDir string) error { - cfg, err := ReadConfig(filepath.Join(dataDir, MachineConfigPath)) - if err != nil { - return err - } - fmt.Println("Config", cfg) - - return errors.New("Running daemon...") -} diff --git a/internal/machine/install.go b/internal/machine/install.go deleted file mode 100644 index 8cc176d5..00000000 --- a/internal/machine/install.go +++ /dev/null @@ -1,15 +0,0 @@ -package machine - -import ( - "fmt" - "path/filepath" - "uncloud/internal/machine/daemon" -) - -func Install(dataDir string, cfg daemon.Config) error { - err := cfg.Write(filepath.Join(dataDir, daemon.MachineConfigPath)) - if err == nil { - fmt.Println("Machine config created.") - } - return err -} diff --git a/internal/machine/network/config.go b/internal/machine/network/config.go index ca9f4388..054f45b0 100644 --- a/internal/machine/network/config.go +++ b/internal/machine/network/config.go @@ -19,7 +19,7 @@ type Config struct { Subnet netip.Prefix PrivateKey secret.Secret PublicKey secret.Secret - Peers []*PeerConfig + Peers []PeerConfig } type PeerConfig struct { diff --git a/internal/machine/network/ip.go b/internal/machine/network/ip.go new file mode 100644 index 00000000..dd83eaa6 --- /dev/null +++ b/internal/machine/network/ip.go @@ -0,0 +1,18 @@ +package network + +import ( + "net" + "net/netip" +) + +// MachineIP returns the IP address of the machine which is the first address in the subnet. +func MachineIP(subnet netip.Prefix) netip.Addr { + return subnet.Masked().Addr().Next() +} + +func prefixToIPNet(prefix netip.Prefix) net.IPNet { + return net.IPNet{ + IP: prefix.Addr().AsSlice(), + Mask: net.CIDRMask(prefix.Bits(), prefix.Addr().BitLen()), + } +} diff --git a/internal/machine/network/wireguard.go b/internal/machine/network/wireguard.go new file mode 100644 index 00000000..a5173273 --- /dev/null +++ b/internal/machine/network/wireguard.go @@ -0,0 +1,27 @@ +package network + +import ( + "fmt" + "golang.zx2c4.com/wireguard/wgctrl/wgtypes" + "time" + "uncloud/internal/secret" +) + +const ( + WireGuardInterfaceName = "uncloud" + WireGuardPort = 51820 + // WireGuardKeepaliveInterval is sensible interval that works with a wide variety of firewalls is 25 seconds. + WireGuardKeepaliveInterval = 25 * time.Second +) + +// NewMachineKeys generates a new WireGuard private and public key pair. +func NewMachineKeys() (privKey, pubKey secret.Secret, err error) { + wgPrivKey, err := wgtypes.GeneratePrivateKey() + if err != nil { + return nil, nil, fmt.Errorf("generate WireGuard private key: %w", err) + } + privKey = wgPrivKey[:] + wgPubKey := wgPrivKey.PublicKey() + pubKey = wgPubKey[:] + return +} diff --git a/internal/machine/network/wireguard_darwin.go b/internal/machine/network/wireguard_darwin.go new file mode 100644 index 00000000..8b60a512 --- /dev/null +++ b/internal/machine/network/wireguard_darwin.go @@ -0,0 +1,22 @@ +//go:build darwin + +package network + +import ( + "context" + "errors" +) + +type WireGuardNetwork struct{} + +func NewWireGuardNetwork() (*WireGuardNetwork, error) { + return &WireGuardNetwork{}, nil +} + +func (n *WireGuardNetwork) Configure(config Config) error { + return errors.New("not implemented on darwin") +} + +func (n *WireGuardNetwork) Run(ctx context.Context) error { + return errors.New("not implemented on darwin") +} diff --git a/internal/machine/network/wireguard_linux.go b/internal/machine/network/wireguard_linux.go new file mode 100644 index 00000000..a354e70f --- /dev/null +++ b/internal/machine/network/wireguard_linux.go @@ -0,0 +1,195 @@ +//go:build linux + +package network + +import ( + "context" + "errors" + "fmt" + "github.com/vishvananda/netlink" + "golang.org/x/sys/unix" + "golang.zx2c4.com/wireguard/wgctrl" + "log/slog" + "net/netip" + "time" +) + +type WireGuardNetwork struct { + link netlink.Link + peers []peer +} + +type peer struct { + config PeerConfig + lastEndpointChangeTime time.Time +} + +func NewWireGuardNetwork() (*WireGuardNetwork, error) { + link, err := createOrGetLink(WireGuardInterfaceName) + if err != nil { + return nil, fmt.Errorf("create or get WireGuard link %q: %v", WireGuardInterfaceName, err) + } + 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. +func createOrGetLink(name string) (netlink.Link, error) { + link, err := netlink.LinkByName(name) + if err == nil { + slog.Info("Found existing WireGuard interface.", "name", name) + return link, nil + } + //goland:noinspection GoTypeAssertionOnErrors + if _, ok := err.(netlink.LinkNotFoundError); !ok { + return nil, fmt.Errorf("find WireGuard link %q: %v", name, err) + } + link = &netlink.GenericLink{ + // TODO: figure out how to set the most appropriate MTU. + LinkAttrs: netlink.LinkAttrs{Name: name}, + LinkType: "wireguard", + } + if err = netlink.LinkAdd(link); err != nil { + return nil, fmt.Errorf("create WireGuard link %q: %v", name, err) + } + slog.Info("Created WireGuard interface.", "name", name) + + // Refetch the link to get the most up-to-date information. + link, err = netlink.LinkByName(name) + if err != nil { + return nil, fmt.Errorf("find created WireGuard link %q: %v", name, err) + } + return link, nil +} + +// Configure applies the given configuration to the WireGuard network interface. +// It updates device and peers settings, subnet, and peer routes. +func (n *WireGuardNetwork) Configure(config Config) error { + // Reconstruct the list of peers, ensuring that the last endpoint change time is preserved for any existing peers. + existingPeersByPublicKey := map[string]peer{} + for _, p := range n.peers { + existingPeersByPublicKey[p.config.PublicKey.String()] = p + } + n.peers = make([]peer, len(config.Peers)) + for i, peerConfig := range config.Peers { + n.peers[i] = peer{ + config: peerConfig, + } + existingPeer, ok := existingPeersByPublicKey[peerConfig.PublicKey.String()] + if ok && existingPeer.config.Endpoint == peerConfig.Endpoint { + n.peers[i].lastEndpointChangeTime = existingPeer.lastEndpointChangeTime + } else { + n.peers[i].lastEndpointChangeTime = time.Now() + } + } + + wg, err := wgctrl.New() + if err != nil { + return fmt.Errorf("create WireGuard client: %w", err) + } + //goland:noinspection GoUnhandledErrorResult + defer wg.Close() + + wgConfig, err := config.toDeviceConfig() + 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) + } + slog.Info("Configured WireGuard interface.", "name", n.link.Attrs().Name) + + if err = n.updateSubnet(config.Subnet); err != nil { + return err + } + slog.Info("Updated the subnet of the WireGuard interface.", + "name", n.link.Attrs().Name, "subnet", config.Subnet) + + // Bring the WireGuard interface up if it's not already up. + if n.link.Attrs().Flags&unix.IFF_UP != unix.IFF_UP { + if err = netlink.LinkSetUp(n.link); err != nil { + return fmt.Errorf("set WireGuard link %q up: %w", n.link.Attrs().Name, err) + } + slog.Info("Brought WireGuard interface up.", "name", n.link.Attrs().Name) + } + if err = n.updatePeerRoutes(); err != nil { + return err + } + slog.Info("Updated routes to peers via the WireGuard interface.", + "name", n.link.Attrs().Name, "peers", len(n.peers)) + + return nil +} + +// updateSubnet assigns the subnet and the first IP address in it to the WireGuard interface. +// It also removes any other addresses that have been added out of band. +func (n *WireGuardNetwork) updateSubnet(subnet netip.Prefix) error { + machineIP := MachineIP(subnet) + ipSubnet := prefixToIPNet(netip.PrefixFrom(machineIP, subnet.Bits())) + if err := netlink.AddrAdd(n.link, &netlink.Addr{IPNet: &ipSubnet}); err != nil { + if !errors.Is(err, unix.EEXIST) { + return fmt.Errorf("add subnet address to WireGuard link %q: %w", n.link.Attrs().Name, err) + } + } + // Remove the old subnet address if it has changed and remove any other addresses that have been added out of band. + linkAddrs, err := netlink.AddrList(n.link, netlink.FAMILY_ALL) + if err != nil { + return fmt.Errorf("list addresses on WireGuard link %q: %w", n.link.Attrs().Name, err) + } + for _, addr := range linkAddrs { + if addr.IPNet.String() == ipSubnet.String() { + continue + } + if err = netlink.AddrDel(n.link, &addr); err != nil { + return fmt.Errorf("remove address %q from WireGuard link %q: %w", addr.IPNet, n.link.Attrs().Name, err) + } + } + return nil +} + +// updatePeerRoutes adds routes to the peers via the WireGuard interface and removes old routes to peers +// that are no longer in the configuration. +func (n *WireGuardNetwork) updatePeerRoutes() error { + // Add routes to the peers via the WireGuard link. + for _, p := range n.peers { + dst := prefixToIPNet(p.config.Subnet) + if err := netlink.RouteAdd(&netlink.Route{ + LinkIndex: n.link.Attrs().Index, + Scope: netlink.SCOPE_LINK, + Dst: &dst, + }); err != nil && !errors.Is(err, unix.EEXIST) { + return fmt.Errorf("add route to WireGuard link %q: %w", n.link.Attrs().Name, err) + } + slog.Debug("Added route to peer via WireGuard interface.", + "name", n.link.Attrs().Name, "peer", dst) + } + // Remove old routes to peers that are no longer in the configuration. + routes, err := netlink.RouteList(n.link, netlink.FAMILY_ALL) + if err != nil { + return fmt.Errorf("list routes on WireGuard link %q: %w", n.link.Attrs().Name, err) + } + for _, route := range routes { + old := true + for _, p := range n.peers { + if route.Dst.String() == p.config.Subnet.String() { + old = false + break + } + } + if old { + if err = netlink.RouteDel(&route); err != nil { + return fmt.Errorf("remove route %q from WireGuard link %q: %w", route.Dst, n.link.Attrs().Name, err) + } + slog.Debug("Removed route to peer via WireGuard interface.", + "name", n.link.Attrs().Name, "peer", route.Dst) + } + } + return nil +} + +func (n *WireGuardNetwork) Run(ctx context.Context) error { + <-ctx.Done() + return nil +}