diff --git a/cmd/uncloud/machine/init.go b/cmd/uncloud/machine/init.go index 5249614b..b6f49035 100644 --- a/cmd/uncloud/machine/init.go +++ b/cmd/uncloud/machine/init.go @@ -22,7 +22,7 @@ func NewInitCommand() *cobra.Command { opts := initOptions{} cmd := &cobra.Command{ Use: "init", - Short: "Initialise a new cluster that consists of the local or remote machine", + Short: "Initialise a new cluster that consists of the local or remote machine.", RunE: func(cmd *cobra.Command, args []string) error { netPrefix, err := netip.ParsePrefix(opts.network) if err != nil { diff --git a/cmd/uncloud/machine/root.go b/cmd/uncloud/machine/root.go index 12313a07..0bf7e4a0 100644 --- a/cmd/uncloud/machine/root.go +++ b/cmd/uncloud/machine/root.go @@ -12,6 +12,7 @@ func NewRootCommand() *cobra.Command { cmd.AddCommand( NewAddCommand(), NewInitCommand(), + NewTokenCommand(), ) return cmd } diff --git a/cmd/uncloud/machine/token.go b/cmd/uncloud/machine/token.go new file mode 100644 index 00000000..912e650d --- /dev/null +++ b/cmd/uncloud/machine/token.go @@ -0,0 +1,38 @@ +package machine + +import ( + "fmt" + "github.com/spf13/cobra" + "uncloud/internal/machine" + "uncloud/internal/machine/daemon" +) + +type tokenOptions struct { + dataDir string +} + +func NewTokenCommand() *cobra.Command { + opts := tokenOptions{} + cmd := &cobra.Command{ + Use: "token", + Short: "Print the local machine's token for adding it to a cluster.", + RunE: func(cmd *cobra.Command, args []string) error { + token, err := daemon.MachineToken(opts.dataDir) + if err != nil { + return fmt.Errorf("get machine token: %w", err) + } + tokenStr, err := token.String() + if err != nil { + return fmt.Errorf("encode machine token: %w", err) + } + fmt.Println(tokenStr) + return nil + }, + } + + cmd.Flags().StringVarP(&opts.dataDir, "data-dir", "d", machine.DefaultDataDir, + "Directory for storing persistent machine state") + _ = cmd.MarkFlagDirname("data-dir") + + return cmd +} diff --git a/internal/machine/config.go b/internal/machine/config.go index 2f46ea1a..2c792ab2 100644 --- a/internal/machine/config.go +++ b/internal/machine/config.go @@ -41,12 +41,16 @@ func ConfigPath(dataDir string) string { func ParseConfig(path string) (*Config, error) { data, err := os.ReadFile(path) if err != nil { - return nil, fmt.Errorf("read config file %q: %w", path, err) + return nil, fmt.Errorf("read config file: %w", err) } var config Config if err = json.Unmarshal(data, &config); err != nil { return nil, fmt.Errorf("parse config file %q: %w", path, err) } + + if config.Network == nil { + return nil, fmt.Errorf("missing network configuration in config file %q", path) + } return &config, nil } diff --git a/internal/machine/daemon/daemon.go b/internal/machine/daemon/daemon.go index 9b61ed76..0aab4b61 100644 --- a/internal/machine/daemon/daemon.go +++ b/internal/machine/daemon/daemon.go @@ -38,7 +38,7 @@ func InitCluster(dataDir, machineName string, netPrefix netip.Prefix, users []*p } // Use all routable addresses as endpoints. - addrs, err := network.ListRoutableAddresses() + addrs, err := network.ListRoutableIPs() if err != nil { return fmt.Errorf("list routable addresses: %w", err) } @@ -112,9 +112,30 @@ type Daemon struct { } func New(dataDir string) (*Daemon, error) { - cfg, err := machine.ParseConfig(machine.ConfigPath(dataDir)) + cfgPath := machine.ConfigPath(dataDir) + cfg, err := machine.ParseConfig(cfgPath) if err != nil { - return nil, fmt.Errorf("load machine config: %w", err) + if !errors.Is(err, os.ErrNotExist) { + return nil, fmt.Errorf("load machine config: %w", err) + } + // Generate an empty machine config with a new key pair. + slog.Info("Machine config not found, creating a new one.", "path", cfgPath) + privKey, pubKey, kErr := network.NewMachineKeys() + if kErr != nil { + return nil, fmt.Errorf("generate machine keys: %w", kErr) + } + slog.Info("Generated machine key pair.", "pubkey", pubKey) + + cfg = &machine.Config{ + Network: &network.Config{ + PrivateKey: privKey, + PublicKey: pubKey, + }, + } + cfg.SetPath(cfgPath) + if err = cfg.Save(); err != nil { + return nil, fmt.Errorf("save machine config: %w", err) + } } statePath := cluster.StatePath(dataDir) @@ -123,59 +144,74 @@ func New(dataDir string) (*Daemon, error) { if !errors.Is(err, os.ErrNotExist) { return nil, fmt.Errorf("load cluster state: %w", err) } - slog.Info("No cluster state found, creating a new one.", "path", statePath) + slog.Info("Cluster state not found, creating a new one.", "path", statePath) if err = state.Save(); err != nil { return nil, fmt.Errorf("save cluster state: %w", err) } } - apiAddr := net.JoinHostPort(cfg.Network.ManagementIP.String(), strconv.Itoa(machine.APIPort)) - c := cluster.NewCluster(state, apiAddr) + d := &Daemon{ + config: cfg, + } + if cfg.Network.IsConfigured() { + apiAddr := net.JoinHostPort(cfg.Network.ManagementIP.String(), strconv.Itoa(machine.APIPort)) + d.cluster = cluster.NewCluster(state, apiAddr) + } - return &Daemon{ - config: cfg, - cluster: c, - }, nil + return d, nil } func (d *Daemon) Run(ctx context.Context) error { - wgnet, err := network.NewWireGuardNetwork() - if err != nil { - return fmt.Errorf("create WireGuard network: %w", err) - } - if err = wgnet.Configure(*d.config.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) - // Use an errgroup to coordinate error handling and graceful shutdown of multiple daemon components. errGroup, ctx := errgroup.WithContext(ctx) - errGroup.Go(func() error { - slog.Info("Starting cluster.") - if err = d.cluster.Run(); err != nil { - return fmt.Errorf("cluster failed: %w", err) + + // Start the network only if it is configured. + if d.config.Network.IsConfigured() { + wgnet, err := network.NewWireGuardNetwork() + if err != nil { + return fmt.Errorf("create WireGuard network: %w", err) } - return nil - }) - errGroup.Go(func() error { - if err = wgnet.Run(ctx); err != nil { - return fmt.Errorf("WireGuard network failed: %w", err) + if err = wgnet.Configure(*d.config.Network); err != nil { + return fmt.Errorf("configure WireGuard network: %w", err) } - return nil - }) + + //ctx, cancel := context.WithCancel(context.Background()) + //go wgnet.WatchEndpoints(ctx, peerEndpointChangeNotifier) + + //addrs, err := network.ListRoutableIPs() + //if err != nil { + // return err + //} + //fmt.Println("Addresses:", addrs) + + errGroup.Go(func() error { + if err = wgnet.Run(ctx); err != nil { + return fmt.Errorf("WireGuard network failed: %w", err) + } + return nil + }) + } else { + slog.Info("Waiting for network configuration to start WireGuard network.") + } + + if d.cluster != nil { + errGroup.Go(func() error { + slog.Info("Starting cluster.") + if err := d.cluster.Run(); err != nil { + return fmt.Errorf("cluster failed: %w", err) + } + return nil + }) + } + // Shutdown goroutine. errGroup.Go(func() error { <-ctx.Done() - slog.Info("Stopping cluster.") - d.cluster.Stop() - slog.Info("Cluster stopped.") + if d.cluster != nil { + slog.Info("Stopping cluster.") + d.cluster.Stop() + slog.Info("Cluster stopped.") + } return nil }) diff --git a/internal/machine/daemon/token.go b/internal/machine/daemon/token.go new file mode 100644 index 00000000..109967eb --- /dev/null +++ b/internal/machine/daemon/token.go @@ -0,0 +1,41 @@ +package daemon + +import ( + "errors" + "fmt" + "net/netip" + "os" + "uncloud/internal/machine" + "uncloud/internal/machine/network" +) + +// MachineToken returns the local machine's token that can be used for adding the machine to a cluster. +// TODO: ideally, this should be an RPC call to the daemon API to ensure the config is created and up-to-date. +func MachineToken(dataDir string) (machine.Token, error) { + cfg, err := machine.ParseConfig(machine.ConfigPath(dataDir)) + if err != nil { + if errors.Is(err, os.ErrNotExist) { + return machine.Token{}, fmt.Errorf("load machine config (is uncloudd daemon running?): %w", err) + } + return machine.Token{}, fmt.Errorf("load machine config: %w", err) + } + if len(cfg.Network.PublicKey) == 0 { + return machine.Token{}, errors.New("public key is not set in machine config") + } + + ips, err := network.ListRoutableIPs() + if err != nil { + return machine.Token{}, fmt.Errorf("list routable addresses: %w", err) + } + publicIP, err := network.GetPublicIP() + // Ignore the error if failed to get the public IP using API services. + if err == nil { + ips = append([]netip.Addr{publicIP}, ips...) + } + + endpoints := make([]netip.AddrPort, len(ips)) + for i, ip := range ips { + endpoints[i] = netip.AddrPortFrom(ip, network.WireGuardPort) + } + return machine.NewToken(cfg.Network.PublicKey, endpoints), nil +} diff --git a/internal/machine/network/address.go b/internal/machine/network/address.go new file mode 100644 index 00000000..68053d9a --- /dev/null +++ b/internal/machine/network/address.go @@ -0,0 +1,106 @@ +package network + +import ( + "context" + "fmt" + "io" + "net" + "net/http" + "net/netip" + "strings" + "time" +) + +// ListRoutableIPs returns a list of routable unicast IP addresses. +func ListRoutableIPs() ([]netip.Addr, error) { + interfaces, err := net.Interfaces() + if err != nil { + return nil, fmt.Errorf("list network interfaces: %w", err) + } + + var routable []netip.Addr + for _, iface := range interfaces { + if iface.Name == WireGuardInterfaceName || strings.HasPrefix(iface.Name, "docker") { + // Skip the Uncloud WireGuard and Docker interfaces. + continue + } + if iface.Flags&net.FlagUp == 0 || iface.Flags&net.FlagRunning == 0 || iface.Flags&net.FlagLoopback != 0 { + // Skip interfaces: + // * Not administratively UP. + // * The operational status is not RUNNING. This is the closest equivalent to checking for NO-CARRIER. + // * Loopback. + continue + } + // TODO: check for link/ether ifaces? + + addrs, aErr := iface.Addrs() + if aErr != nil { + return nil, fmt.Errorf("list unicast addresses for interface %q: %w", iface.Name, err) + } + for _, addr := range addrs { + ipNet, ok := addr.(*net.IPNet) + if !ok { + continue + } + // Includes IPv4 private address space and local IPv6 unicast address space. + if ipNet.IP.IsGlobalUnicast() { + ip, pErr := netip.ParseAddr(ipNet.IP.String()) + if pErr != nil { + return nil, fmt.Errorf("parse IP address %q: %w", ipNet.IP, err) + } + routable = append(routable, ip) + } + } + } + return routable, nil +} + +func GetPublicIP() (netip.Addr, error) { + services := []struct { + URL string + Parser func([]byte) (netip.Addr, error) + }{ + {"https://api.ipify.org", parsePlaintextIP}, + {"https://ipinfo.io/ip", parsePlaintextIP}, + {"http://ip-api.com/line/?fields=query", parsePlaintextIP}, + } + + for _, service := range services { + if ip, err := queryIP(service.URL, service.Parser); err == nil { + return ip, nil + } + } + + return netip.Addr{}, fmt.Errorf("failed to get public IP from all services") +} + +func queryIP(service string, parser func([]byte) (netip.Addr, error)) (netip.Addr, error) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, service, nil) + if err != nil { + return netip.Addr{}, fmt.Errorf("create request: %w", err) + } + + resp, err := http.DefaultClient.Do(req) + if err != nil { + return netip.Addr{}, fmt.Errorf("send request: %w", err) + } + defer func() { + _ = resp.Body.Close() + }() + if resp.StatusCode != http.StatusOK { + return netip.Addr{}, fmt.Errorf("unexpected status code: %d", resp.StatusCode) + } + + body, err := io.ReadAll(resp.Body) + if err != nil { + return netip.Addr{}, fmt.Errorf("read response body: %w", err) + } + return parser(body) +} + +func parsePlaintextIP(data []byte) (netip.Addr, error) { + return netip.ParseAddr(string(data)) +} diff --git a/internal/machine/network/config.go b/internal/machine/network/config.go index 939caaa9..bec519d5 100644 --- a/internal/machine/network/config.go +++ b/internal/machine/network/config.go @@ -35,6 +35,11 @@ type PeerConfig struct { PublicKey secret.Secret } +func (c Config) IsConfigured() bool { + return c.Subnet != (netip.Prefix{}) && c.ManagementIP != (netip.Addr{}) && + c.PrivateKey != nil && c.PublicKey != nil +} + func (c Config) toDeviceConfig() (wgtypes.Config, error) { privateKey, err := wgtypes.NewKey(c.PrivateKey) if err != nil { diff --git a/internal/machine/token.go b/internal/machine/token.go new file mode 100644 index 00000000..1bbc529c --- /dev/null +++ b/internal/machine/token.go @@ -0,0 +1,54 @@ +package machine + +import ( + "encoding/base64" + "encoding/json" + "fmt" + "net/netip" + "strings" + "uncloud/internal/secret" +) + +const ( + TokenPrefix = "mtkn:" +) + +// Token represents the machine's token for joining a cluster. +type Token struct { + PublicKey secret.Secret + Endpoints []netip.AddrPort +} + +// NewToken creates a new machine token with the given public key and endpoints. +func NewToken(publicKey secret.Secret, endpoints []netip.AddrPort) Token { + return Token{ + PublicKey: publicKey, + Endpoints: endpoints, + } +} + +// ParseToken decodes a machine token from the given string. +func ParseToken(s string) (Token, error) { + if strings.HasPrefix(s, TokenPrefix) { + return Token{}, fmt.Errorf("invalid token prefix") + } + decoded, err := base64.StdEncoding.DecodeString(s[len(TokenPrefix):]) + if err != nil { + return Token{}, fmt.Errorf("decode token: %w", err) + } + var token Token + if err = json.Unmarshal(decoded, &token); err != nil { + return Token{}, fmt.Errorf("unmarshal token: %w", err) + } + return token, nil +} + +// String returns the machine token encoded as a string. +func (t Token) String() (string, error) { + js, err := json.Marshal(t) + if err != nil { + return "", fmt.Errorf("marshal token: %w", err) + } + encoded := base64.StdEncoding.EncodeToString(js) + return TokenPrefix + encoded, nil +}