kilo/cmd/kgctl/connect_linux.nogo

// +build linux

package main

import (
	"context"
	"errors"
	"fmt"
	"io/ioutil"
	"net"
	"os"
	"strings"
	"syscall"
	"time"

	"github.com/go-kit/kit/log"
	"github.com/go-kit/kit/log/level"
	"github.com/oklog/run"
	"github.com/spf13/cobra"
	"github.com/vishvananda/netlink"
	"golang.org/x/sys/unix"
	"golang.zx2c4.com/wireguard/wgctrl"
	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/tools/clientcmd"

	"github.com/squat/kilo/pkg/iproute"
	"github.com/squat/kilo/pkg/k8s"
	"github.com/squat/kilo/pkg/k8s/apis/kilo/v1alpha1"
	kiloclient "github.com/squat/kilo/pkg/k8s/clientset/versioned"
	"github.com/squat/kilo/pkg/mesh"
	"github.com/squat/kilo/pkg/route"
	"github.com/squat/kilo/pkg/wireguard"
)

func takeIPNet(_ net.IP, i *net.IPNet, _ error) *net.IPNet {
	return i
}

func connect() *cobra.Command {
	cmd := &cobra.Command{
		Use:   "connect",
		RunE:  connectAsPeer,
		Short: "connect to a Kilo cluster as a peer over WireGuard",
	}
	cmd.Flags().IPNetP("allowed-ip", "a", *takeIPNet(net.ParseCIDR("10.10.10.10/32")), "Allowed IP of the peer")
	cmd.Flags().IPNetP("service-cidr", "c", *takeIPNet(net.ParseCIDR("10.43.0.0/16")), "service CIDR of the cluster")
	cmd.Flags().String("log-level", logLevelInfo, fmt.Sprintf("Log level to use. Possible values: %s", availableLogLevels))
	cmd.Flags().String("config-path", "/tmp/wg.ini", "path to WireGuard configuation file")
	cmd.Flags().Bool("clean-up", true, "clean up routes and interface")
	cmd.Flags().Uint("mtu", uint(1420), "clean up routes and interface")
	cmd.Flags().Duration("resync-period", 30*time.Second, "How often should Kilo reconcile?")

	availableLogLevels = strings.Join([]string{
		logLevelAll,
		logLevelDebug,
		logLevelInfo,
		logLevelWarn,
		logLevelError,
		logLevelNone,
	}, ", ")

	return cmd
}

func connectAsPeer(cmd *cobra.Command, args []string) error {
	var cancel context.CancelFunc
	resyncPersiod, err := cmd.Flags().GetDuration("resync-period")
	if err != nil {
		return err
	}
	mtu, err := cmd.Flags().GetUint("mtu")
	if err != nil {
		return err
	}
	configPath, err := cmd.Flags().GetString("config-path")
	if err != nil {
		return err
	}
	serviceCIDR, err := cmd.Flags().GetIPNet("service-cidr")
	if err != nil {
		return err
	}
	allowedIP, err := cmd.Flags().GetIPNet("allowed-ip")
	if err != nil {
		return err
	}
	logger := log.NewJSONLogger(log.NewSyncWriter(os.Stdout))
	logLevel, err := cmd.Flags().GetString("log-level")
	if err != nil {
		return err
	}
	switch logLevel {
	case logLevelAll:
		logger = level.NewFilter(logger, level.AllowAll())
	case logLevelDebug:
		logger = level.NewFilter(logger, level.AllowDebug())
	case logLevelInfo:
		logger = level.NewFilter(logger, level.AllowInfo())
	case logLevelWarn:
		logger = level.NewFilter(logger, level.AllowWarn())
	case logLevelError:
		logger = level.NewFilter(logger, level.AllowError())
	case logLevelNone:
		logger = level.NewFilter(logger, level.AllowNone())
	default:
		return fmt.Errorf("log level %s unknown; possible values are: %s", logLevel, availableLogLevels)
	}
	logger = log.With(logger, "ts", log.DefaultTimestampUTC)
	logger = log.With(logger, "caller", log.DefaultCaller)
	peername := "random"
	if len(args) > 0 {
		peername = args[0]
	}

	var kiloClient *kiloclient.Clientset
	switch backend {
	case k8s.Backend:
		config, err := clientcmd.BuildConfigFromFlags("", kubeconfig)
		if err != nil {
			return fmt.Errorf("failed to create Kubernetes config: %v", err)
		}
		kiloClient = kiloclient.NewForConfigOrDie(config)
	default:
		return fmt.Errorf("backend %v unknown; posible values are: %s", backend, availableBackends)
	}
	privateKey, err := wgtypes.GeneratePrivateKey()
	if err != nil {
		return fmt.Errorf("failed to generate private key: %w", err)
	}
	publicKey := privateKey.PublicKey()
	level.Info(logger).Log("msg", "generated public key", "key", publicKey)

	peer := &v1alpha1.Peer{
		ObjectMeta: metav1.ObjectMeta{
			Name: peername,
		},
		Spec: v1alpha1.PeerSpec{
			AllowedIPs:          []string{allowedIP.String()},
			PersistentKeepalive: 10,
			PublicKey:           publicKey.String(),
		},
	}
	if p, err := kiloClient.KiloV1alpha1().Peers().Get(context.TODO(), peername, metav1.GetOptions{}); err != nil || p == nil {
		peer, err = kiloClient.KiloV1alpha1().Peers().Create(context.TODO(), peer, metav1.CreateOptions{})
		if err != nil {
			return fmt.Errorf("failed to create peer: %w", err)
		}
	}

	kiloIfaceName := "kilo0"

	iface, _, err := wireguard.New(kiloIfaceName, mtu)
	if err != nil {
		return fmt.Errorf("failed to create wg interface: %w", err)
	} else {
		level.Info(logger).Log("msg", "successfully created wg interface", "name", kiloIfaceName, "no", iface)
	}
	if err := iproute.Set(iface, false); err != nil {
		return err
	}

	if err := iproute.SetAddress(iface, &allowedIP); err != nil {
		return err
	} else {
		level.Info(logger).Log("mag", "successfully set IP address of wg interface", "IP", allowedIP.String())
	}

	if err := iproute.Set(iface, true); err != nil {
		return err
	}

	var g run.Group
	ctx, cancel := context.WithCancel(context.Background())
	g.Add(run.SignalHandler(ctx, syscall.SIGINT, syscall.SIGTERM))

	table := route.NewTable()
	stop := make(chan struct{}, 1)
	errCh := make(<-chan error, 1)
	{
		ch := make(chan struct{})
		g.Add(
			func() error {
				for {
					select {
					case err, ok := <-errCh:
						if ok {
							level.Error(logger).Log("err", err.Error())
						} else {
							return nil
						}
					case <-ch:
						return nil
					}
				}
			},
			func(err error) {
				ch <- struct{}{}
				close(ch)
				stop <- struct{}{}
				close(stop)
				level.Error(logger).Log("msg", "stopped ip routes table", "err", err.Error())
			},
		)
	}
	{
		ch := make(chan struct{})
		g.Add(
			func() error {
				for {
					ns, err := opts.backend.Nodes().List()
					if err != nil {
						return fmt.Errorf("failed to list nodes: %v", err)
					}
					ps, err := opts.backend.Peers().List()
					if err != nil {
						return fmt.Errorf("failed to list peers: %v", err)
					}
					// Obtain the Granularity by looking at the annotation of the first node.
					if opts.granularity, err = optainGranularity(opts.granularity, ns); err != nil {
						return fmt.Errorf("failed to obtain granularity: %w", err)
					}
					var hostname string
					subnet := mesh.DefaultKiloSubnet
					nodes := make(map[string]*mesh.Node)
					for _, n := range ns {
						if n.Ready() {
							nodes[n.Name] = n
							hostname = n.Name
						}
						if n.WireGuardIP != nil {
							subnet = n.WireGuardIP
						}
					}
					subnet.IP = subnet.IP.Mask(subnet.Mask)
					if len(nodes) == 0 {
						return errors.New("did not find any valid Kilo nodes in the cluster")
					}
					peers := make(map[string]*mesh.Peer)
					for _, p := range ps {
						if p.Ready() {
							peers[p.Name] = p
						}
					}
					if _, ok := peers[peername]; !ok {
						return fmt.Errorf("did not find any peer named %q in the cluster", peername)
					}

					t, err := mesh.NewTopology(nodes, peers, opts.granularity, hostname, opts.port, []byte{}, subnet, peers[peername].PersistentKeepalive, logger)
					if err != nil {
						return fmt.Errorf("failed to create topology: %v", err)
					}
					conf := t.PeerConf(peername)
					conf.Interface = &wireguard.Interface{
						PrivateKey: []byte(privateKey.String()),
						ListenPort: uint32(55555),
					}
					buf, err := conf.Bytes()
					if err != nil {
						//level.Error(m.logger).Log("error", err)
						//m.errorCounter.WithLabelValues("apply").Inc()
						return err
					}
					if err := ioutil.WriteFile("/tmp/wg.ini", buf, 0600); err != nil {
						//level.Error(m.logger).Log("error", err)
						//m.errorCounter.WithLabelValues("apply").Inc()
						return err
					}
					if err := wireguard.SetConf(kiloIfaceName, "/tmp/wg.ini"); err != nil {
						return err
					}
					wgClient, err := wgctrl.New()
					if err != nil {
						return fmt.Errorf("failed to initialize wg Client: %w", err)
					}
					defer func() {
						wgClient.Close()
					}()
					wgConf := wgtypes.Config{
						PrivateKey: &privateKey,
						//	Peers:      []wgtypes.PeerConfig{},
					}
					if err := wgClient.ConfigureDevice(kiloIfaceName, wgConf); err != nil {
						return fmt.Errorf("failed to configure wg interface: %w", err)
					}

					var routes []*netlink.Route
					for _, segment := range t.Segments() {
						for i := range segment.CIDRS() {
							// Add routes to the Pod CIDRs of nodes in other segments.
							routes = append(routes, &netlink.Route{
								Dst: segment.CIDRS()[i],
								//Flags:     int(netlink.FLAG_ONLINK),
								LinkIndex: iface,
								Protocol:  unix.RTPROT_STATIC,
							})
						}
						for i := range segment.PrivateIPs() {
							// Add routes to the private IPs of nodes in other segments.
							routes = append(routes, &netlink.Route{
								Dst: mesh.OneAddressCIDR(segment.PrivateIPs()[i]),
								//Flags:     int(netlink.FLAG_ONLINK),
								LinkIndex: iface,
								Protocol:  unix.RTPROT_STATIC,
							})
						}
						// For segments / locations other than the location of this instance of kg,
						// we need to set routes for allowed location IPs over the leader in the current location.
						for i := range segment.AllowedLocationIPs() {
							routes = append(routes, &netlink.Route{
								Dst: segment.AllowedLocationIPs()[i],
								//Flags:     int(netlink.FLAG_ONLINK),
								LinkIndex: iface,
								Protocol:  unix.RTPROT_STATIC,
							})
						}
						routes = append(routes, &netlink.Route{
							Dst: mesh.OneAddressCIDR(segment.WireGuardIP()),
							//Flags:     int(netlink.FLAG_ONLINK),
							LinkIndex: iface,
							Protocol:  unix.RTPROT_STATIC,
						})
					}
					// Add routes for the allowed IPs of peers.
					for _, peer := range t.Peers() {
						for i := range peer.AllowedIPs {
							routes = append(routes, &netlink.Route{
								Dst: peer.AllowedIPs[i],
								//Flags:     int(netlink.FLAG_ONLINK),
								LinkIndex: iface,
								Protocol:  unix.RTPROT_STATIC,
							})
						}
					}
					routes = append(routes, &netlink.Route{
						Dst: &serviceCIDR,
						//Flags:     int(netlink.FLAG_ONLINK),
						Gw:        nil,
						LinkIndex: iface,
						Protocol:  unix.RTPROT_STATIC,
					})
					for _, r := range routes {
						fmt.Println(r)
					}

					level.Debug(logger).Log("routes", routes)
					if err := table.Set(routes, []*netlink.Rule{}); err != nil {
						return fmt.Errorf("failed to set ip routes table: %w", err)
					}
					errCh, err = table.Run(stop)
					if err != nil {
						return fmt.Errorf("failed to start ip routes tables: %w", err)
					}
					select {
					case <-time.After(resyncPersiod):
					case <-ch:
						return nil
					}
				}
			}, func(err error) {
				// Cancel the root context in the very end.
				defer func() {
					cancel()
					level.Debug(logger).Log("msg", "canceled parent context")
				}()
				ch <- struct{}{}
				var serr run.SignalError
				if ok := errors.As(err, &serr); ok {
					level.Info(logger).Log("msg", "received signal", "signal", serr.Signal.String(), "err", err.Error())
				} else {
					level.Error(logger).Log("msg", "received error", "err", err.Error())
				}
				level.Debug(logger).Log("msg", "stoped ip routes table")
				ctxWithTimeOut, cancelWithTimeOut := context.WithTimeout(ctx, 10*time.Second)
				defer func() {
					cancelWithTimeOut()
					level.Debug(logger).Log("msg", "canceled timed context")
				}()
				if err := kiloClient.KiloV1alpha1().Peers().Delete(ctxWithTimeOut, peername, metav1.DeleteOptions{}); err != nil {
					level.Error(logger).Log("failed to delete peer: %w", err)
				} else {
					level.Info(logger).Log("msg", "deleted peer", "peer", peername)
				}
				if ok, err := cmd.Flags().GetBool("clean-up"); err != nil {
					level.Error(logger).Log("err", err.Error(), "msg", "failed to get value from clean-up flag")
				} else if ok {
					cleanUp(iface, table, configPath, logger)
				}
			})
	}
	err = g.Run()
	var serr run.SignalError
	if ok := errors.As(err, &serr); ok {
		return nil
	}
	return err
}

func cleanUp(iface int, t *route.Table, configPath string, logger log.Logger) {
	if err := iproute.Set(iface, false); err != nil {
		level.Error(logger).Log("err", err.Error(), "msg", "failed to set down wg interface")
	}
	if err := os.Remove(configPath); err != nil {
		level.Error(logger).Log("error", fmt.Sprintf("failed to delete configuration file: %v", err))
	}
	if err := iproute.RemoveInterface(iface); err != nil {
		level.Error(logger).Log("error", fmt.Sprintf("failed to remove WireGuard interface: %v", err))
	}
	if err := t.CleanUp(); err != nil {
		level.Error(logger).Log("failed to clean up routes: %w", err)
	}
	return
}

//func nodeToWGPeer(n *mesh.Node) (ret wgtypes.PeerConfig, err error) {
//	pubKey, err := wgtypes.ParseKey(string(n.Key))
//	if err != nil {
//		return ret, err
//	}
//	ret.PublicKey = pubKey
//	if err != nil {
//		return ret, err
//	}
//	aIPs := []net.IPNet{*n.WireGuardIP, *n.InternalIP, *n.Subnet}
//	for _, a := range n.AllowedLocationIPs {
//		aIPs = append(aIPs, *a)
//	}
//	ret.AllowedIPs = aIPs
//
//	dur := time.Second * time.Duration(n.PersistentKeepalive)
//	ret.PersistentKeepaliveInterval = &dur
//
//	udpEndpoint, err := net.ResolveUDPAddr("udp", n.Endpoint.String())
//	if err != nil {
//		return ret, err
//	} else if udpEndpoint.IP == nil {
//		udpEndpoint = nil
//	}
//	ret.Endpoint = udpEndpoint
//	return ret, nil
//}
//
//func toWGPeer(p wireguard.Peer) (ret wgtypes.PeerConfig, err error) {
//	pubKey, err := wgtypes.ParseKey(string(p.PublicKey))
//	if err != nil {
//		return ret, err
//	}
//	ret.PublicKey = pubKey
//	aIPs := make([]net.IPNet, len(p.AllowedIPs))
//	for i, a := range p.AllowedIPs {
//		aIPs[i] = *a
//	}
//	ret.AllowedIPs = aIPs
//
//	if preSharedKey, err := wgtypes.ParseKey(string(p.PresharedKey)); len(p.PresharedKey) > 0 && err != nil {
//		return ret, err
//	} else {
//		ret.PresharedKey = &preSharedKey
//	}
//	dur := time.Second * time.Duration(p.PersistentKeepalive)
//	ret.PersistentKeepaliveInterval = &dur
//
//	udpEndpoint, err := net.ResolveUDPAddr("udp", p.Endpoint.String())
//	if err != nil {
//		return ret, err
//	} else if udpEndpoint.IP == nil {
//		udpEndpoint = nil
//	}
//	fmt.Println("peer")
//	fmt.Println(aIPs)
//	ret.Endpoint = udpEndpoint
//	return ret, nil
//}