From 9e44d303289b0d38ce98468598b6441c6a875b79 Mon Sep 17 00:00:00 2001 From: Pavel Sviderski Date: Tue, 10 Sep 2024 13:48:14 +1000 Subject: [PATCH] implement CLI init cluster, add SSH and WireGuard cluster connectors --- internal/cli/cli.go | 225 ++++++++++++++++ internal/cli/client/client.go | 41 +++ internal/cli/client/cluster.go | 294 +++++++++++++++++++++ internal/cli/client/connector/ssh.go | 79 ++++++ internal/cli/client/connector/wireguard.go | 83 ++++++ internal/cli/{ => client}/user.go | 4 +- internal/cli/cluster.go | 213 --------------- internal/cli/machine.go | 13 +- 8 files changed, 729 insertions(+), 223 deletions(-) create mode 100644 internal/cli/client/client.go create mode 100644 internal/cli/client/cluster.go create mode 100644 internal/cli/client/connector/ssh.go create mode 100644 internal/cli/client/connector/wireguard.go rename internal/cli/{ => client}/user.go (93%) delete mode 100644 internal/cli/cluster.go diff --git a/internal/cli/cli.go b/internal/cli/cli.go index d1fdf1ad..10843d52 100644 --- a/internal/cli/cli.go +++ b/internal/cli/cli.go @@ -1,8 +1,20 @@ package cli import ( + "context" + "crypto/ed25519" + "errors" "fmt" + "net/netip" + "uncloud/internal/cli/client" + "uncloud/internal/cli/client/connector" "uncloud/internal/cli/config" + "uncloud/internal/machine/api/pb" + "uncloud/internal/secret" +) + +var ( + ErrNotFound = errors.New("not found") ) type CLI struct { @@ -18,3 +30,216 @@ func New(configPath string) (*CLI, error) { config: cfg, }, nil } + +func (cli *CLI) CreateCluster( + name string, privateKey ed25519.PrivateKey, userPrivateKey secret.Secret, +) (*client.ClusterClient, error) { + if _, ok := cli.config.Clusters[name]; ok { + return nil, fmt.Errorf("cluster %q already exists", name) + } + if privateKey == nil { + var err error + _, privateKey, err = ed25519.GenerateKey(nil) + if err != nil { + return nil, fmt.Errorf("generate cluster secret: %w", err) + } + } + if userPrivateKey == nil { + user, err := client.NewUser(nil) + if err != nil { + return nil, fmt.Errorf("generate user: %w", err) + } + userPrivateKey = user.PrivateKey() + } + + cli.config.Clusters[name] = &config.Cluster{ + Name: name, + Secret: privateKey.Seed(), + UserPrivateKey: userPrivateKey, + } + if err := cli.config.Save(); err != nil { + return nil, err + } + + return cli.GetCluster(name) +} + +func (cli *CLI) CreateDefaultCluster() (*client.ClusterClient, error) { + c, err := cli.CreateCluster("default", nil, nil) + if err != nil { + return nil, err + } + if err = cli.SetCurrentCluster(c.Name()); err != nil { + return nil, err + } + return c, nil +} + +func (cli *CLI) GetCluster(name string) (*client.ClusterClient, error) { + clusterCfg, ok := cli.config.Clusters[name] + if !ok { + return nil, ErrNotFound + } + clusterCfg.Name = name + + user, err := client.NewUser(clusterCfg.UserPrivateKey) + if err != nil { + return nil, fmt.Errorf("create user: %w", err) + } + wgConnector := connector.NewWireGuardConnector(user, clusterCfg.Machines) + + return client.NewClusterClient(clusterCfg, wgConnector) +} + +func (cli *CLI) GetCurrentCluster() (*client.ClusterClient, error) { + return cli.GetCluster(cli.config.CurrentCluster) +} + +func (cli *CLI) SetCurrentCluster(name string) error { + if _, ok := cli.config.Clusters[name]; !ok { + return ErrNotFound + } + cli.config.CurrentCluster = name + return cli.config.Save() +} + +func (cli *CLI) ListClusters() ([]*client.ClusterClient, error) { + var clusters []*client.ClusterClient + for name := range cli.config.Clusters { + c, err := cli.GetCluster(name) + if err != nil { + return nil, fmt.Errorf("get cluster %q: %w", name, err) + } + clusters = append(clusters, c) + } + return clusters, nil +} + +func (cli *CLI) InitCluster( + ctx context.Context, remoteMachine *RemoteMachine, clusterName, machineName string, netPrefix netip.Prefix, +) error { + if remoteMachine != nil { + return cli.initRemoteMachine(ctx, *remoteMachine, clusterName, machineName, netPrefix) + } + // TODO: implement local machine initialisation + return fmt.Errorf("local machine initialisation is not implemented yet") +} + +func (cli *CLI) initRemoteMachine( + ctx context.Context, remoteMachine RemoteMachine, clusterName, machineName string, netPrefix netip.Prefix, +) error { + sshConfig := &connector.SSHConnectorConfig{ + User: remoteMachine.User, + Host: remoteMachine.Host, + Port: remoteMachine.Port, + KeyPath: remoteMachine.KeyPath, + } + c, err := client.New(ctx, connector.NewSSHConnector(sshConfig)) + if err != nil { + return fmt.Errorf("connect to remote machine: %w", err) + } + defer func() { + _ = c.Close() + }() + + var cluster *client.ClusterClient + if clusterName == "" { + cluster, err = cli.CreateDefaultCluster() + } else { + cluster, err = cli.CreateCluster(clusterName, nil, nil) + } + if err != nil { + return err + } + user, err := cluster.User() + if err != nil { + return fmt.Errorf("get cluster user: %w", err) + } + + // TODO: download and install the latest uncloudd binary by running the install shell script from GitHub. + // For now upload the binary using scp manually. + // TODO: Check if the machine is already provisioned and ask the user to reset it first. + + req := &pb.InitClusterRequest{ + MachineName: machineName, + Network: pb.NewIPPrefix(netPrefix), + User: &pb.User{ + Network: &pb.NetworkConfig{ + ManagementIp: pb.NewIP(user.ManagementIP()), + PublicKey: user.PublicKey(), + }, + }, + } + resp, err := c.InitCluster(ctx, req) + if err != nil { + return fmt.Errorf("init cluster: %w", err) + } + fmt.Printf("Cluster %q initialised with machine %q\n", cluster.Name(), resp.Machine.Name) + + // Save the machine's SSH connection details in the cluster config. + connCfg := config.MachineConnection{ + SSH: config.NewSSHDestination(remoteMachine.User, remoteMachine.Host, remoteMachine.Port), + } + cli.config.Clusters[cluster.Name()].Machines = append(cli.config.Clusters[cluster.Name()].Machines, connCfg) + if err = cli.config.Save(); err != nil { + return fmt.Errorf("save config: %w", err) + } + return nil +} + +func (cli *CLI) AddMachine( + ctx context.Context, clusterName, machineName, user, host string, port int, sshKeyPath string, +) error { + var ( + cluster *client.ClusterClient + err error + ) + if clusterName == "" { + // If the cluster is not specified, use the current cluster. If there are no clusters, create a default one. + cluster, err = cli.GetCurrentCluster() + if err != nil { + if errors.Is(err, ErrNotFound) { + // Do not create a default cluster if there are already clusters but the current cluster is not set. + clusters, cErr := cli.ListClusters() + if cErr != nil { + return fmt.Errorf("list clusters: %w", cErr) + } + if len(clusters) > 0 { + return errors.New( + "the current cluster is not set in the Uncloud config. " + + "Please specify a cluster with the --cluster flag or set current_cluster in the config", + ) + } + + cluster, err = cli.CreateDefaultCluster() + if err != nil { + return fmt.Errorf("create default cluster: %w", err) + } + fmt.Printf("Created %q cluster\n", cluster.Name()) + } else { + return fmt.Errorf("get current cluster: %w", err) + } + } + } else { + cluster, err = cli.GetCluster(clusterName) + if err != nil { + return fmt.Errorf("get cluster %q: %w", clusterName, err) + } + } + defer func() { + _ = cluster.Close() + }() + + name, connCfg, err := cluster.AddMachine(ctx, machineName, user, host, port, sshKeyPath) + if err != nil { + return fmt.Errorf("add machine to cluster %q: %w", cluster.Name(), err) + } + fmt.Printf("Machine %q added to cluster %q\n", name, cluster.Name()) + + cli.config.Clusters[cluster.Name()].Machines = append(cli.config.Clusters[cluster.Name()].Machines, connCfg) + if err = cli.config.Save(); err != nil { + return fmt.Errorf("save config: %w", err) + } + + return nil +} diff --git a/internal/cli/client/client.go b/internal/cli/client/client.go new file mode 100644 index 00000000..1e225bf0 --- /dev/null +++ b/internal/cli/client/client.go @@ -0,0 +1,41 @@ +package client + +import ( + "context" + "errors" + "fmt" + "google.golang.org/grpc" + "uncloud/internal/machine/api/pb" +) + +type Client struct { + connector Connector + conn *grpc.ClientConn + + pb.MachineClient +} + +// Connector is an interface for establishing a connection to the machine API. +type Connector interface { + Connect(ctx context.Context) (*grpc.ClientConn, error) + Close() error +} + +func New(ctx context.Context, connector Connector) (*Client, error) { + c := &Client{ + connector: connector, + } + var err error + c.conn, err = connector.Connect(ctx) + if err != nil { + return nil, fmt.Errorf("connect to machine: %w", err) + } + + c.MachineClient = pb.NewMachineClient(c.conn) + return c, nil +} + +func (c *Client) Close() error { + err := c.conn.Close() + return errors.Join(err, c.connector.Close()) +} diff --git a/internal/cli/client/cluster.go b/internal/cli/client/cluster.go new file mode 100644 index 00000000..a959567f --- /dev/null +++ b/internal/cli/client/cluster.go @@ -0,0 +1,294 @@ +package client + +import ( + "context" + "crypto/ed25519" + "errors" + "fmt" + "google.golang.org/grpc" + "net/netip" + "uncloud/internal/cli/config" + "uncloud/internal/machine" + "uncloud/internal/machine/api/pb" + "uncloud/internal/machine/network" + "uncloud/internal/secret" + "uncloud/internal/sshexec" +) + +type ClusterClient struct { + config *config.Cluster + + connector Connector + conn *grpc.ClientConn + client pb.ClusterClient +} + +func NewClusterClient(cfg *config.Cluster, connector Connector) (*ClusterClient, error) { + if cfg.UserPrivateKey == nil { + return nil, errors.New("cluster user_key must be set in the config") + } + return &ClusterClient{ + config: cfg, + connector: connector, + }, nil +} + +func (c *ClusterClient) Name() string { + return c.config.Name +} + +// HasMachines returns true if the cluster has at least one machine specified in the config. +func (c *ClusterClient) HasMachines() bool { + return len(c.config.Machines) > 0 +} + +func (c *ClusterClient) User() (*User, error) { + return NewUser(c.config.UserPrivateKey) +} + +// TODO: implement Connect method that establishes a WireGuard tunnel to a cluster machine +// +// and initializes an API client through it. +func (c *ClusterClient) connect(ctx context.Context) error { + if c.conn != nil { + return nil + } + if !c.HasMachines() { + return errors.New("no machines specified in the cluster config") + } + + conn, err := c.connector.Connect(ctx) + if err != nil { + return fmt.Errorf("connect to cluster: %w", err) + } + c.conn = conn + c.client = pb.NewClusterClient(conn) + + return nil +} + +func (c *ClusterClient) Close() error { + c.connector.Close() + if c.conn != nil { + err := c.conn.Close() + c.conn = nil + c.client = nil + return err + } + return nil +} + +func (c *ClusterClient) AddMachine( + ctx context.Context, name, user, host string, port int, sshKeyPath string, +) (string, config.MachineConnection, error) { + client, err := sshexec.Connect(user, host, port, sshKeyPath) + if err != nil { + return "", config.MachineConnection{}, fmt.Errorf("SSH login to %s@%s:%d: %w", user, host, port, err) + } + exec := sshexec.NewRemote(client) + defer func() { + _ = exec.Close() + }() + + // TODO: download and install the latest uncloudd binary by running the install shell script from GitHub. + // For now upload the binary using scp manually. + // TODO: Check if the machine is already provisioned and ask the user to reset it first. + // TODO: grab a list of routable IP addresses from the remote machine. + + addrs := []netip.Addr{} + + sudoPrefix := "" + if user != "root" { + sudoPrefix = "sudo" + } + + if !c.HasMachines() { + clusterUser, uErr := NewUser(c.config.UserPrivateKey) + if uErr != nil { + return "", config.MachineConnection{}, uErr + } + + _, rErr := exec.Run(ctx, sshexec.QuoteCommand( + sudoPrefix, "uncloud", "machine", "init", + "--name", name, + "--user-pubkey", clusterUser.PublicKey().String())) + if rErr != nil { + return "", config.MachineConnection{}, fmt.Errorf("initialise a new cluster on machine: %w", rErr) + } + } else { + endpoints := make([]*pb.IPPort, len(addrs)) + for i, addr := range addrs { + addrPort := netip.AddrPortFrom(addr, network.WireGuardPort) + endpoints[i] = pb.NewIPPort(addrPort) + } + + mcfg, err := c.newMachineConfig(ctx, name, addrs) + if err != nil { + return "", config.MachineConnection{}, fmt.Errorf("create machine config: %w", err) + } + + _, err = exec.Run(ctx, sshexec.QuoteCommand(sudoPrefix, "mkdir", "-m", "700", "-p", machine.DefaultDataDir)) + if err != nil { + return "", config.MachineConnection{}, fmt.Errorf("create data directory %q: %w", machine.DefaultDataDir, err) + } + + // Write the machine config to /var/lib/uncloud/machine.json by piping the JSON data to the file. + mcfgData, err := mcfg.Encode() + if err != nil { + return "", config.MachineConnection{}, fmt.Errorf("encode machine config: %w", err) + } + mcfgPath := sshexec.Quote(machine.StatePath(machine.DefaultDataDir)) + createFileCmd := fmt.Sprintf("%s touch %s && %s chmod 600 %s", sudoPrefix, mcfgPath, sudoPrefix, mcfgPath) + _, err = exec.Run(ctx, fmt.Sprintf("%s && echo %s | %s tee %s > /dev/null", + createFileCmd, sshexec.Quote(string(mcfgData)), sudoPrefix, mcfgPath)) + if err != nil { + return "", config.MachineConnection{}, fmt.Errorf("write machine config to %q: %w", mcfgPath, err) + } + fmt.Println("Machine config written to", mcfgPath) + } + + out, err := exec.Run(ctx, sshexec.QuoteCommand(sudoPrefix, "systemctl", "restart", "uncloudd")) + if err != nil { + return "", config.MachineConnection{}, fmt.Errorf("start uncloudd: %w: %s", err, out) + } + + // Get the machine token to retrieve the public key from it. + tokenOut, err := exec.Run(ctx, sshexec.QuoteCommand(sudoPrefix, "uncloud", "machine", "token")) + if err != nil { + return "", config.MachineConnection{}, fmt.Errorf("get machine token: %w: %s", err, out) + } + token, err := machine.ParseToken(tokenOut) + if err != nil { + return "", config.MachineConnection{}, fmt.Errorf("parse machine token: %w", err) + } + // TODO: replace command runs with sending gRPC request to the machine API via unix socket. + name, err = exec.Run(ctx, fmt.Sprintf("%s cat %s | grep Name | cut -d'\"' -f4", + sudoPrefix, machine.StatePath(machine.DefaultDataDir))) + if err != nil { + return "", config.MachineConnection{}, fmt.Errorf("get machine name: %w: %s", err, out) + } + + connCfg := config.MachineConnection{ + Host: host, + PublicKey: token.PublicKey, + } + return name, connCfg, nil +} + +// newMachineConfig creates a new machine config for a machine that is being added to the cluster. +// addrs is a list of routable IP addresses that the machine can be reached at. +func (c *ClusterClient) newMachineConfig(ctx context.Context, name string, addrs []netip.Addr) (*machine.State, error) { + if !c.HasMachines() { + // Create a bootstrap config for the first machine in the cluster. + clusterUser, err := NewUser(c.config.UserPrivateKey) + if err != nil { + return nil, err + } + userPeerCfg := network.PeerConfig{ + ManagementIP: clusterUser.ManagementIP(), + PublicKey: clusterUser.PublicKey(), + } + mcfg, err := machine.NewBootstrapConfig(name, netip.Prefix{}, userPeerCfg) + if err != nil { + return nil, fmt.Errorf("generate machine bootstrap config: %w", err) + } + return mcfg, nil + } + + // Create a config for a new machine in the cluster that has already been bootstrapped. + privKey, pubKey, err := network.NewMachineKeys() + if err != nil { + return nil, fmt.Errorf("generate machine keys: %w", err) + } + endpoints := make([]netip.AddrPort, len(addrs)) + for i, addr := range addrs { + // Hardcode the WireGuard port until it's required to be configurable. + endpoints[i] = netip.AddrPortFrom(addr, network.WireGuardPort) + } + resp, err := c.registerNewMachine(ctx, name, endpoints, pubKey) + if err != nil { + return nil, fmt.Errorf("register new machine: %w", err) + } + minfo := resp.Machine + fmt.Printf("Machine %q registered in the cluster with ID %q\n", minfo.Name, minfo.Id) + + //peers := make([]network.PeerConfig, len(resp.OtherMachines)) + //for i, pinfo := range resp.OtherMachines { + // peer := pinfo.Network + // if len(peer.Endpoints) == 0 { + // continue + // } + // peerSubnet, pErr := pinfo.Network.Subnet.ToPrefix() + // if pErr != nil { + // return nil, pErr + // } + // peerManageIP, pErr := pinfo.Network.ManagementIp.ToAddr() + // if pErr != nil { + // return nil, pErr + // } + // peerEndpoints := make([]netip.AddrPort, len(peer.Endpoints)) + // for j, ep := range peer.Endpoints { + // if peerEndpoints[j], err = ep.ToAddrPort(); err != nil { + // return nil, pErr + // } + // } + // peers[i] = network.PeerConfig{ + // Subnet: &peerSubnet, + // ManagementIP: peerManageIP, + // // TODO: do not pick an endpoint and let the daemon do it. + // Endpoint: &peerEndpoints[0], + // AllEndpoints: peerEndpoints, + // PublicKey: pinfo.Network.PublicKey, + // } + //} + + subnet, err := minfo.Network.Subnet.ToPrefix() + if err != nil { + return nil, err + } + manageIP, err := minfo.Network.ManagementIp.ToAddr() + if err != nil { + return nil, err + } + mcfg := &machine.State{ + ID: minfo.Id, + Name: minfo.Name, + Network: &network.Config{ + Subnet: subnet, + ManagementIP: manageIP, + PrivateKey: privKey, + PublicKey: pubKey, + //Peers: peers, + }, + } + return mcfg, nil +} + +func (c *ClusterClient) registerNewMachine( + ctx context.Context, name string, endpoints []netip.AddrPort, publicKey secret.Secret, +) (*pb.AddMachineResponse, error) { + if err := c.connect(ctx); err != nil { + return nil, err + } + + pbEndpoints := make([]*pb.IPPort, len(endpoints)) + for i, ep := range endpoints { + pbEndpoints[i] = pb.NewIPPort(ep) + } + req := &pb.AddMachineRequest{ + Name: name, + Network: &pb.NetworkConfig{ + Endpoints: pbEndpoints, + PublicKey: publicKey, + }, + } + return c.client.AddMachine(ctx, req) +} + +func privateKeyFromSecret(s secret.Secret) (ed25519.PrivateKey, error) { + // Cluster secret in the config is a hex-encoded private key seed. + if len(s) != ed25519.SeedSize { + return nil, fmt.Errorf("invalid cluster secret length") + } + return ed25519.NewKeyFromSeed(s), nil +} diff --git a/internal/cli/client/connector/ssh.go b/internal/cli/client/connector/ssh.go new file mode 100644 index 00000000..fcaa4ad3 --- /dev/null +++ b/internal/cli/client/connector/ssh.go @@ -0,0 +1,79 @@ +package connector + +import ( + "context" + "fmt" + "golang.org/x/crypto/ssh" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + "net" + "strings" + "uncloud/internal/machine" + "uncloud/internal/sshexec" +) + +type SSHConnectorConfig struct { + User string + Host string + Port int + KeyPath string + + APISockPath string +} + +// SSHConnector establishes a connection to the machine API through an SSH tunnel to the machine. +type SSHConnector struct { + config SSHConnectorConfig + client *ssh.Client +} + +func NewSSHConnector(cfg *SSHConnectorConfig) *SSHConnector { + c := &SSHConnector{config: *cfg} + if c.config.User == "" { + c.config.User = "root" + } + if c.config.Port == 0 { + c.config.Port = 22 + } + if c.config.APISockPath == "" { + c.config.APISockPath = machine.DefaultAPISockPath + } + return c +} + +// TODO: handle context cancelation. +func (c *SSHConnector) Connect(ctx context.Context) (*grpc.ClientConn, error) { + var err error + c.client, err = sshexec.Connect(c.config.User, c.config.Host, c.config.Port, c.config.KeyPath) + if err != nil { + return nil, fmt.Errorf("SSH login to %s@%s:%d: %w", c.config.User, c.config.Host, c.config.Port, err) + } + + conn, err := grpc.NewClient( + "unix://"+c.config.APISockPath, + grpc.WithTransportCredentials(insecure.NewCredentials()), + grpc.WithContextDialer( + func(ctx context.Context, addr string) (net.Conn, error) { + addr = strings.TrimPrefix(addr, "unix://") + conn, dErr := c.client.DialContext(ctx, "unix", addr) + if dErr != nil { + return nil, fmt.Errorf("connect to machine API socket %s through SSH tunnel: %w", addr, dErr) + } + return conn, nil + }, + ), + ) + if err != nil { + return nil, fmt.Errorf("create machine API client: %w", err) + } + return conn, nil +} + +func (c *SSHConnector) Close() error { + if c.client != nil { + err := c.client.Close() + c.client = nil + return err + } + return nil +} diff --git a/internal/cli/client/connector/wireguard.go b/internal/cli/client/connector/wireguard.go new file mode 100644 index 00000000..0b27af91 --- /dev/null +++ b/internal/cli/client/connector/wireguard.go @@ -0,0 +1,83 @@ +package connector + +import ( + "context" + "fmt" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + "net" + "net/netip" + "strconv" + "uncloud/internal/cli/client" + "uncloud/internal/cli/config" + machine2 "uncloud/internal/machine" + "uncloud/internal/machine/network" + "uncloud/internal/machine/network/tunnel" +) + +// WireGuardConnector establishes a connection to the cluster API through a WireGuard tunnel +// to one of the cluster machines. +type WireGuardConnector struct { + user *client.User + machines []config.MachineConnection + tun *tunnel.Tunnel +} + +func NewWireGuardConnector(user *client.User, machines []config.MachineConnection) *WireGuardConnector { + return &WireGuardConnector{ + user: user, + machines: machines, + } +} + +// TODO: handle context cancelation. +func (c *WireGuardConnector) Connect(ctx context.Context) (*grpc.ClientConn, error) { + if len(c.machines) == 0 { + return nil, fmt.Errorf("no machines to connect to") + } + // TODO: iterate over machines and try to connect to each one until successful. + // For now, try to connect to only the first machine. + machine := c.machines[0] + endpointIPs, err := net.LookupIP(machine.Host) + if err != nil { + return nil, fmt.Errorf("resolve IP for %q: %w", machine.Host, err) + } + endpointAddr, err := netip.ParseAddr(endpointIPs[0].String()) + if err != nil { + return nil, fmt.Errorf("parse IP address %q: %w", endpointIPs[0].String(), err) + } + endpoint := netip.AddrPortFrom(endpointAddr, tunnel.DefaultEndpointPort) + machineManagementIP := network.ManagementIP(machine.PublicKey) + machineAPIAddr := net.JoinHostPort(machineManagementIP.String(), strconv.Itoa(machine2.APIPort)) + + tunCfg := &tunnel.Config{ + LocalAddress: c.user.ManagementIP(), + LocalPrivateKey: c.user.PrivateKey(), + RemotePublicKey: machine.PublicKey, + RemoteNetwork: netip.PrefixFrom(machineManagementIP, 128), + Endpoint: endpoint, + } + if c.tun, err = tunnel.Connect(tunCfg); err != nil { + return nil, fmt.Errorf("establish WireGuard tunnel to %q: %w", endpoint, err) + } + + conn, err := grpc.NewClient( + machineAPIAddr, + grpc.WithTransportCredentials(insecure.NewCredentials()), + grpc.WithContextDialer(func(ctx context.Context, addr string) (net.Conn, error) { + return c.tun.DialContext(ctx, "tcp", addr) + }), + ) + if err != nil { + return nil, fmt.Errorf("connect to machine API through WireGuard tunnel: %w", err) + } + return conn, nil +} + +func (c *WireGuardConnector) Close() error { + if c.tun != nil { + c.tun.Close() + c.tun = nil + } + return nil +} diff --git a/internal/cli/user.go b/internal/cli/client/user.go similarity index 93% rename from internal/cli/user.go rename to internal/cli/client/user.go index b33f85c5..b267d464 100644 --- a/internal/cli/user.go +++ b/internal/cli/client/user.go @@ -1,4 +1,4 @@ -package cli +package client import ( "fmt" @@ -45,5 +45,5 @@ func (u *User) PublicKey() secret.Secret { } func (u *User) ManagementIP() netip.Addr { - return network.PeerIPv6(u.PublicKey()) + return network.ManagementIP(u.PublicKey()) } diff --git a/internal/cli/cluster.go b/internal/cli/cluster.go deleted file mode 100644 index aba00de8..00000000 --- a/internal/cli/cluster.go +++ /dev/null @@ -1,213 +0,0 @@ -package cli - -import ( - "context" - "crypto/ed25519" - "errors" - "fmt" - "net/netip" - "uncloud/internal/cli/config" - "uncloud/internal/cmdexec" - "uncloud/internal/machine" - "uncloud/internal/machine/network" - "uncloud/internal/secret" -) - -var ( - ErrNotFound = errors.New("not found") -) - -type Cluster struct { - Name string - privateKey ed25519.PrivateKey - - config *config.Config -} - -func (c *Cluster) User() (*User, error) { - userKey := c.config.Clusters[c.Name].UserKey - if userKey == nil { - return nil, errors.New("cluster user_key must be set in the config") - } - return NewUser(userKey) -} - -func (c *Cluster) Machines() []config.MachineConnection { - cfg, ok := c.config.Clusters[c.Name] - if !ok { - return nil - } - return cfg.Machines -} - -func (c *Cluster) toConfig() *config.Cluster { - return &config.Cluster{ - Name: c.Name, - Secret: c.privateKey.Seed(), - } -} - -func (cli *CLI) CreateCluster(name string, privateKey ed25519.PrivateKey, userPrivateKey secret.Secret) (*Cluster, error) { - if _, ok := cli.config.Clusters[name]; ok { - return nil, fmt.Errorf("cluster %q already exists", name) - } - if privateKey == nil { - var err error - _, privateKey, err = ed25519.GenerateKey(nil) - if err != nil { - return nil, fmt.Errorf("generate cluster secret: %w", err) - } - } - if userPrivateKey == nil { - user, err := NewUser(nil) - if err != nil { - return nil, fmt.Errorf("generate user: %w", err) - } - userPrivateKey = user.PrivateKey() - } - - c := &Cluster{ - Name: name, - privateKey: privateKey, - config: cli.config, - } - cfg := c.toConfig() - cfg.UserKey = userPrivateKey - cli.config.Clusters[name] = cfg - if err := cli.config.Save(); err != nil { - return nil, err - } - - return c, nil -} - -func (cli *CLI) CreateDefaultCluster() (*Cluster, error) { - c, err := cli.CreateCluster("default", nil, nil) - if err != nil { - return nil, err - } - if err = cli.SetCurrentCluster(c.Name); err != nil { - return nil, err - } - return c, nil -} - -func (cli *CLI) GetCluster(name string) (*Cluster, error) { - clusterCfg, ok := cli.config.Clusters[name] - if !ok { - return nil, ErrNotFound - } - privateKey, err := privateKeyFromSecret(clusterCfg.Secret) - if err != nil { - return nil, err - } - - return &Cluster{ - Name: name, - privateKey: privateKey, - config: cli.config, - }, nil -} - -func (cli *CLI) GetCurrentCluster() (*Cluster, error) { - return cli.GetCluster(cli.config.CurrentCluster) -} - -func (cli *CLI) SetCurrentCluster(name string) error { - if _, ok := cli.config.Clusters[name]; !ok { - return ErrNotFound - } - cli.config.CurrentCluster = name - return cli.config.Save() -} - -func (cli *CLI) ListClusters() ([]*Cluster, error) { - var clusters []*Cluster - for name := range cli.config.Clusters { - c, err := cli.GetCluster(name) - if err != nil { - return nil, fmt.Errorf("get cluster %q: %w", name, err) - } - clusters = append(clusters, c) - } - return clusters, nil -} - -func (c *Cluster) AddMachine(ctx context.Context, name, user, host string, port int, sshKeyPath string) (string, error) { - exec, err := cmdexec.Connect(user, host, port, sshKeyPath) - if err != nil { - return "", fmt.Errorf("SSH login to %s@%s:%d: %w", user, host, port, err) - } - defer func() { - _ = exec.Close() - }() - - clusterUser, err := c.User() - if err != nil { - return "", err - } - userPeerCfg := network.PeerConfig{ - ManagementIP: clusterUser.ManagementIP(), - PublicKey: clusterUser.PublicKey(), - } - mcfg, err := machine.NewBootstrapConfig(name, netip.Prefix{}, userPeerCfg) - if err != nil { - return "", fmt.Errorf("generate machine bootstrap config: %w", err) - } - - sudoPrefix := "" - if user != "root" { - sudoPrefix = "sudo" - } - - _, err = exec.Run(ctx, cmdexec.QuoteCommand(sudoPrefix, "mkdir", "-p", machine.DefaultDataDir)) - if err != nil { - return "", fmt.Errorf("create data directory %q: %w", machine.DefaultDataDir, err) - } - - // TODO: Check if the machine is already provisioned and ask the user to reset it first. - // Write the machine config to /var/lib/uncloud/machine.json by piping the JSON data to the file. - mcfgData, err := mcfg.Encode() - if err != nil { - return "", fmt.Errorf("encode machine config: %w", err) - } - mcfgPath := cmdexec.Quote(machine.ConfigPath(machine.DefaultDataDir)) - createFileCmd := fmt.Sprintf("%s touch %s && %s chmod 600 %s", sudoPrefix, mcfgPath, sudoPrefix, mcfgPath) - _, err = exec.Run(ctx, fmt.Sprintf("%s && echo %s | %s tee %s > /dev/null", - createFileCmd, cmdexec.Quote(string(mcfgData)), sudoPrefix, mcfgPath)) - 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 daemon started") - - connConfig := config.MachineConnection{ - User: user, - Host: host, - Port: port, - SSHKey: sshKeyPath, - PublicKey: mcfg.Network.PublicKey, - } - c.config.Clusters[c.Name].Machines = append(c.config.Clusters[c.Name].Machines, connConfig) - if err = c.config.Save(); err != nil { - return "", fmt.Errorf("save config: %w", err) - } - - return mcfg.Name, nil -} - -func privateKeyFromSecret(s secret.Secret) (ed25519.PrivateKey, error) { - // Cluster secret in the config is a hex-encoded private key seed. - if len(s) != ed25519.SeedSize { - return nil, fmt.Errorf("invalid cluster secret length") - } - return ed25519.NewKeyFromSeed(s), nil -} diff --git a/internal/cli/machine.go b/internal/cli/machine.go index 731a9e89..3345266e 100644 --- a/internal/cli/machine.go +++ b/internal/cli/machine.go @@ -1,11 +1,8 @@ package cli -import "uncloud/internal/cli/config" - -type Machine struct { - connConfig config.MachineConnection -} - -func NewMachine(connConfig config.MachineConnection) *Machine { - return &Machine{connConfig: connConfig} +type RemoteMachine struct { + User string + Host string + Port int + KeyPath string }