mirror of
https://github.com/psviderski/uncloud.git
synced 2026-08-26 19:13:34 +00:00
implement CLI init cluster, add SSH and WireGuard cluster connectors
This commit is contained in:
Pavel Sviderski
@@ -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())
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -0,0 +1,49 @@
|
||||
package client
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
|
||||
"net/netip"
|
||||
"uncloud/internal/machine/network"
|
||||
"uncloud/internal/secret"
|
||||
)
|
||||
|
||||
type User struct {
|
||||
privateKey wgtypes.Key
|
||||
}
|
||||
|
||||
func NewUser(privateKey secret.Secret) (*User, error) {
|
||||
var (
|
||||
wgKey wgtypes.Key
|
||||
err error
|
||||
)
|
||||
if privateKey == nil {
|
||||
wgKey, err = wgtypes.GeneratePrivateKey()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("generate key for user: %w", err)
|
||||
}
|
||||
privateKey = wgKey[:]
|
||||
} else {
|
||||
wgKey, err = wgtypes.NewKey(privateKey)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("invalid key: %w", err)
|
||||
}
|
||||
privateKey = wgKey[:]
|
||||
}
|
||||
return &User{
|
||||
privateKey: wgKey,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (u *User) PrivateKey() secret.Secret {
|
||||
return u.privateKey[:]
|
||||
}
|
||||
|
||||
func (u *User) PublicKey() secret.Secret {
|
||||
pubKey := u.privateKey.PublicKey()
|
||||
return pubKey[:]
|
||||
}
|
||||
|
||||
func (u *User) ManagementIP() netip.Addr {
|
||||
return network.ManagementIP(u.PublicKey())
|
||||
}
|
||||
Reference in New Issue
Block a user