Merge branch 'main' of github.com:EdgeNet-project/node

go.mod

@@ -4,15 +4,16 @@ go 1.16
 
 require (
 	cloud.google.com/go v0.94.1
-	github.com/EdgeNet-project/edgenet v1.0.0-alpha.1.0.20210701214804-887efc4a97b0
+	github.com/EdgeNet-project/edgenet v1.0.0-alpha.1.0.20210913151028-60f26ffd5fb3
 	github.com/aws/aws-sdk-go v1.40.35
 	github.com/coreos/go-iptables v0.6.0
 	github.com/scaleway/scaleway-sdk-go v1.0.0-beta.7
 	github.com/thanhpk/randstr v1.0.4
 	github.com/txn2/txeh v1.3.0
-	github.com/vishvananda/netlink v1.1.0
+	github.com/vishvananda/netlink v1.1.1-0.20210530105856-14e832ae1e8f
 	github.com/yumaojun03/dmidecode v0.1.4
 	golang.org/x/sys v0.0.0-20210823070655-63515b42dcdf
+	golang.zx2c4.com/wireguard/wgctrl v0.0.0-20210506160403-92e472f520a5
 	gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b
 	k8s.io/apimachinery v0.22.1
 	k8s.io/client-go v0.22.1

main.go

@@ -21,18 +21,25 @@ import (
 	"github.com/EdgeNet-project/node/pkg/cluster"
 	"github.com/EdgeNet-project/node/pkg/network"
 	"github.com/EdgeNet-project/node/pkg/platforms"
+	"github.com/EdgeNet-project/node/pkg/utils"
 	"github.com/thanhpk/randstr"
+	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
 	"gopkg.in/yaml.v3"
 	"log"
+	"math/rand"
 	"net"
 	"os"
 	"path/filepath"
 	"strings"
+	"time"
 )
 
 const defaultKubeconfigURL = "https://raw.githubusercontent.com/EdgeNet-project/edgenet/master/configs/public.cfg"
+const defaultVPNNetworkV4 = "10.183.0.0/20"
+const defaultVPNNetworkV6 = "fdb4:ae86:ec99:4004::/64"
 const edgenetConfigFile = "/opt/edgenet/config.yaml"
 const kubeletEnvFile = "/etc/default/kubelet"
+const vpnLinkName = "edgenetmesh0"
 
 func check(err error) {
 	if err != nil {
@@ -55,6 +62,16 @@ type edgenetConfig struct {
 	// PublicIPv4 is the public IPv4 address of the host.
 	// This address must be reachable from the Internet.
 	PublicIPv4 net.IP `yaml:"publicIPv4"`
+	// VPNIPv4 is the private IPv4 address of the VPN mesh interface.
+	// This address must be unique among the node in the cluster.
+	VPNIPv4 *utils.IPWithMask `yaml:"vpnIPv4"`
+	// VPNIPv4 is the private IPv6 address of the VPN mesh interface.
+	// This address must be unique among the node in the cluster.
+	VPNIPv6 *utils.IPWithMask `yaml:"vpnIPv6"`
+	// VPNPrivateKey is the WireGuard private key of the VPN mesh interface.
+	VPNPrivateKey string `yaml:"vpnPrivateKey"`
+	// VPNListenPort is the WireGuard port of the VPN mesh interface.
+	VPNListenPort int `yaml:"vpnListenPort"`
 }
 
 // load the EdgeNet configuration from the specified file.
@@ -165,26 +182,62 @@ func main() {
 		config.LocalIPv4, config.PublicIPv4 = getIPv4(config.Platform)
 	}
 
+	// https://github.com/EdgeNet-project/edgenet/issues/156
+	if config.VPNIPv4 == nil || config.VPNIPv6 == nil {
+		log.Println("step=get-vpn-ip")
+		_, vpnNetworkV4, err := net.ParseCIDR(defaultVPNNetworkV4)
+		check(err)
+		_, vpnNetworkV6, err := net.ParseCIDR(defaultVPNNetworkV6)
+		check(err)
+		config.VPNIPv4, config.VPNIPv6 = cluster.FindVPNIPs(defaultKubeconfigURL, *vpnNetworkV4, *vpnNetworkV6)
+	}
+
+	if config.VPNPrivateKey == "" {
+		log.Println("step=get-vpn-private-key")
+		key, err := wgtypes.GeneratePrivateKey()
+		check(err)
+		config.VPNPrivateKey = key.String()
+	}
+
+	if config.VPNListenPort == 0 {
+		log.Println("step=get-vpn-listen-port")
+		rand.Seed(time.Now().UnixNano())
+		config.VPNListenPort = rand.Intn(32768) + 32768
+	}
+
 	log.Println("step=save-config")
 	log.Printf("config=%+v\n", config)
 	config.save(edgenetConfigFile)
 
-	// Cloud providers assign a public IP to instances through NAT.
-	// The instance only sees an private _internal_ IP.
-	// This is problematic for Kubernetes, which expects to see the public IP on the interface.
-	// In this script, we assign the public IP to the instance interface.
-	if !config.LocalIPv4.Equal(config.PublicIPv4) {
-		log.Println("step=set-public-ip")
-		network.AssignPublicIP(config.LocalIPv4, config.PublicIPv4)
-		// This doesn't seems to be required anymore with Antrea (as it was with Calico).
-		// network.RewritePublicIP(config.LocalIPv4, config.PublicIPv4)
-		network.SetKubeletNodeIP(kubeletEnvFile, config.PublicIPv4)
-	}
-
 	log.Println("step=set-hostname")
 	hostname := fmt.Sprintf("%s-%s.edge-net.io", config.HostnameRoot, config.HostnameSuffix)
 	network.SetHostname(hostname)
 
+	// https://github.com/EdgeNet-project/edgenet/issues/156
+	if config.VPNIPv4 != nil && config.VPNIPv6 != nil {
+		log.Println("step=configure-vpn")
+		network.InitializeVPN(vpnLinkName, config.VPNPrivateKey, config.VPNListenPort)
+		network.AssignVPNIP(vpnLinkName, *config.VPNIPv4, *config.VPNIPv6)
+		privateKey, err := wgtypes.ParseKey(config.VPNPrivateKey)
+		check(err)
+		cluster.CreateVPNPeer(defaultKubeconfigURL, hostname, config.PublicIPv4, config.VPNIPv4.IP, config.VPNIPv6.IP, config.VPNListenPort, privateKey.PublicKey().String())
+		// Pre-establish the tunnels before the VPNPeer controller gets started.
+		peers := cluster.ListVPNPeer(defaultKubeconfigURL)
+		for _, peer := range peers {
+			network.AddPeer(vpnLinkName, peer)
+		}
+	}
+
+	var nodeIP net.IP
+	if config.LocalIPv4.Equal(config.PublicIPv4) {
+		nodeIP = config.PublicIPv4
+	} else {
+		nodeIP = config.VPNIPv4.IP
+	}
+
+	log.Println("step=set-node-ip")
+	network.SetKubeletNodeIP(kubeletEnvFile, nodeIP)
+
 	log.Println("step=join-cluster")
-	cluster.Join(defaultKubeconfigURL, config.PublicIPv4, hostname)
+	cluster.Join(defaultKubeconfigURL, hostname, nodeIP)
 }

pkg/cluster/cluster.go

@@ -19,7 +19,9 @@ package cluster
 import (
 	"context"
 	"github.com/EdgeNet-project/edgenet/pkg/apis/core/v1alpha"
+	v1alpha2 "github.com/EdgeNet-project/edgenet/pkg/apis/networking/v1alpha"
 	"github.com/EdgeNet-project/edgenet/pkg/generated/clientset/versioned"
+	"github.com/EdgeNet-project/node/pkg/utils"
 	"io/ioutil"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -49,14 +51,67 @@ func configFromUrl(url string) (*rest.Config, error) {
 	return clientcmd.RESTConfigFromKubeConfig(buf)
 }
 
+func FindVPNIPs(configURL string, netv4 net.IPNet, netv6 net.IPNet) (*utils.IPWithMask, *utils.IPWithMask) {
+	peers := ListVPNPeer(configURL)
+	usedIPs := make([]net.IP, 0)
+	for _, peer := range peers {
+		usedIPs = append(usedIPs, net.ParseIP(peer.Spec.AddressV4))
+	}
+	ipv4 := utils.RandIPv4(netv4, usedIPs)
+	ipv6 := make(net.IP, 16)
+	copy(ipv6[:12], netv6.IP[:12])
+	copy(ipv6[12:16], ipv4)
+	return &utils.IPWithMask{IP: ipv4, Mask: netv4.Mask}, &utils.IPWithMask{IP: ipv6, Mask: netv6.Mask}
+}
+
+func CreateVPNPeer(configURL string, hostname string, externalIP net.IP, ipv4 net.IP, ipv6 net.IP, listenPort int, publicKey string) {
+	config, err := configFromUrl(configURL)
+	check(err)
+	clientset, err := versioned.NewForConfig(config)
+	check(err)
+	client := clientset.NetworkingV1alpha().VPNPeers()
+	_externalIP := externalIP.String()
+	peer := &v1alpha2.VPNPeer{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: hostname,
+		},
+		Spec: v1alpha2.VPNPeerSpec{
+			AddressV4:       ipv4.String(),
+			AddressV6:       ipv6.String(),
+			EndpointAddress: &_externalIP,
+			EndpointPort:    &listenPort,
+			PublicKey:       publicKey,
+		},
+	}
+	_, err = client.Create(context.TODO(), peer.DeepCopy(), metav1.CreateOptions{})
+	if errors.IsAlreadyExists(err) {
+		log.Print("vpn-peer-status=already-exists")
+	} else if err != nil {
+		panic(err)
+	} else {
+		log.Print("vpn-peer-status=created")
+	}
+}
+
+func ListVPNPeer(configURL string) []v1alpha2.VPNPeer {
+	config, err := configFromUrl(configURL)
+	check(err)
+	clientset, err := versioned.NewForConfig(config)
+	check(err)
+	client := clientset.NetworkingV1alpha().VPNPeers()
+	peers, err := client.List(context.TODO(), metav1.ListOptions{})
+	check(err)
+	return peers.Items
+}
+
 // Join the node to the cluster specified by configURL.
 // It ignores AlreadyExists errors when creating the NodeContribution object.
-func Join(configURL string, externalIP net.IP, hostname string) {
+func Join(configURL string, hostname string, externalIP net.IP) {
 	config, err := configFromUrl(configURL)
 	check(err)
 	clientset, err := versioned.NewForConfig(config)
 	check(err)
-	nodeContributionClient := clientset.CoreV1alpha().NodeContributions()
+	client := clientset.CoreV1alpha().NodeContributions()
 	nodeContribution := &v1alpha.NodeContribution{
 		ObjectMeta: metav1.ObjectMeta{
 			Name: strings.ReplaceAll(hostname, ".edge-net.io", ""),
@@ -69,7 +124,7 @@ func Join(configURL string, externalIP net.IP, hostname string) {
 			User:    "edgenet",
 		},
 	}
-	_, err = nodeContributionClient.Create(context.TODO(), nodeContribution.DeepCopy(), metav1.CreateOptions{})
+	_, err = client.Create(context.TODO(), nodeContribution.DeepCopy(), metav1.CreateOptions{})
 	if errors.IsAlreadyExists(err) {
 		log.Print("node-contribution-status=already-exists")
 	} else if err != nil {

pkg/network/vpn.go

@@ -15,3 +15,114 @@ limitations under the License.
 */
 
 package network
+
+import (
+	"github.com/EdgeNet-project/edgenet/pkg/apis/networking/v1alpha"
+	"github.com/EdgeNet-project/node/pkg/utils"
+	"github.com/vishvananda/netlink"
+	"golang.org/x/sys/unix"
+	"golang.zx2c4.com/wireguard/wgctrl"
+	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
+	"net"
+	"time"
+)
+
+func getOrAddVPNLink(name string) netlink.Link {
+	link, err := netlink.LinkByName(name)
+	if err == nil {
+		return link
+	}
+	_, ok := err.(netlink.LinkNotFoundError)
+	if !ok {
+		panic(err)
+	}
+	linkAttrs := netlink.NewLinkAttrs()
+	linkAttrs.Name = name
+	link = &netlink.Wireguard{LinkAttrs: linkAttrs}
+	check(netlink.LinkAdd(link))
+	return link
+}
+
+func InitializeVPN(name string, privateKey string, listenPort int) {
+	link := getOrAddVPNLink(name)
+	check(netlink.LinkSetUp(link))
+	client, err := wgctrl.New()
+	check(err)
+	key, err := wgtypes.ParseKey(privateKey)
+	check(err)
+	config := wgtypes.Config{PrivateKey: &key, ListenPort: &listenPort}
+	check(client.ConfigureDevice(name, config))
+}
+
+func AssignVPNIP(name string, ipv4 utils.IPWithMask, ipv6 utils.IPWithMask) {
+	link := getOrAddVPNLink(name)
+
+	addr4, err := netlink.ParseAddr(ipv4.String())
+	check(err)
+	addrs, err := netlink.AddrList(link, unix.AF_INET)
+	check(err)
+	for _, addr := range addrs {
+		if !addr.Equal(*addr4) {
+			check(netlink.AddrDel(link, &addr))
+		}
+	}
+
+	addr6, err := netlink.ParseAddr(ipv6.String())
+	check(err)
+	addrs, err = netlink.AddrList(link, unix.AF_INET6)
+	check(err)
+	for _, addr := range addrs {
+		if !addr.Equal(*addr6) {
+			check(netlink.AddrDel(link, &addr))
+		}
+	}
+
+	check(netlink.AddrReplace(link, addr4))
+	check(netlink.AddrReplace(link, addr6))
+}
+
+func AddPeer(name string, peer v1alpha.VPNPeer) {
+	client, err := wgctrl.New()
+	check(err)
+
+	publicKey, err := wgtypes.ParseKey(peer.Spec.PublicKey)
+	check(err)
+
+	allowedIPs := []net.IPNet{
+		{
+			IP:   net.ParseIP(peer.Spec.AddressV4),
+			Mask: net.CIDRMask(32, 32),
+		},
+		{
+			IP:   net.ParseIP(peer.Spec.AddressV6),
+			Mask: net.CIDRMask(128, 128),
+		},
+	}
+
+	var endpoint *net.UDPAddr
+	if peer.Spec.EndpointAddress != nil && peer.Spec.EndpointPort != nil {
+		endpoint = &net.UDPAddr{
+			IP:   net.ParseIP(*peer.Spec.EndpointAddress),
+			Port: *peer.Spec.EndpointPort,
+		}
+	}
+
+	keepaliveInterval := 5 * time.Second
+
+	peerConfig := wgtypes.PeerConfig{
+		AllowedIPs:                  allowedIPs,
+		Endpoint:                    endpoint,
+		PublicKey:                   publicKey,
+		PersistentKeepaliveInterval: &keepaliveInterval,
+		Remove:                      false,
+		ReplaceAllowedIPs:           true,
+		UpdateOnly:                  false,
+	}
+
+	deviceConfig := wgtypes.Config{
+		Peers:        []wgtypes.PeerConfig{peerConfig},
+		ReplacePeers: false,
+	}
+
+	check(client.ConfigureDevice(name, deviceConfig))
+}

pkg/platforms/platforms.go

@@ -68,13 +68,12 @@ func Detect() string {
 
 	// GCP
 	log.Printf("try-detect=%s", GCP)
-	_, err = gcpmetadata.InstanceName()
-	if err == nil {
+	name, err := gcpmetadata.InstanceName()
+	if err == nil && name != "" {
 		return GCP
 	}
 
 	// GENI
-
 	if utils.Exists("/usr/local/etc/emulab") {
 		return GENI
 	}
@@ -105,3 +104,13 @@ func Detect() string {
 	// Fallback
 	return Generic
 }
+
+// IsCloud returns whether the platform is a cloud provider or not.
+func IsCloud(platform string) bool {
+	switch platform {
+	case Azure, EC2, GCP, SCW:
+		return true
+	default:
+		return false
+	}
+}