diff --git a/.github/workflows/e2e.yml b/.github/workflows/e2e.yml index a8d4e04d3..b5825165d 100644 --- a/.github/workflows/e2e.yml +++ b/.github/workflows/e2e.yml @@ -44,7 +44,7 @@ jobs: strategy: fail-fast: false matrix: - cable-driver: ['', 'wireguard', 'vxlan'] + cable-driver: ['', 'wireguard', 'amneziawg', 'vxlan'] extra-toggles: ['', 'ovn'] globalnet: ['', 'globalnet'] # Run most tests against the latest K8s version @@ -52,6 +52,8 @@ jobs: exclude: - cable-driver: wireguard extra-toggles: ovn + - cable-driver: amneziawg + extra-toggles: ovn - globalnet: globalnet extra-toggles: ovn cable-driver: diff --git a/.github/workflows/flake_finder.yml b/.github/workflows/flake_finder.yml index f58eaf9ac..de83f84a9 100644 --- a/.github/workflows/flake_finder.yml +++ b/.github/workflows/flake_finder.yml @@ -16,7 +16,7 @@ jobs: strategy: fail-fast: false matrix: - cable-driver: ['', 'wireguard', 'vxlan'] + cable-driver: ['', 'wireguard', 'amneziawg', 'vxlan'] extra-toggles: ['', 'ovn'] globalnet: ['', 'globalnet'] # Run most tests against the latest K8s version @@ -24,6 +24,8 @@ jobs: exclude: - cable-driver: wireguard extra-toggles: ovn + - cable-driver: amneziawg + extra-toggles: ovn include: - extra-toggles: dual-stack - extra-toggles: dual-stack, globalnet diff --git a/go.mod b/go.mod index d29d46300..719486bc2 100644 --- a/go.mod +++ b/go.mod @@ -3,6 +3,8 @@ module github.com/submariner-io/submariner go 1.26.0 require ( + github.com/advanced-wg/awgctrl-go v1.0.0 + github.com/amnezia-vpn/amneziawg-go v0.2.20-0.20260709022233-c1e9bb3758e7 github.com/cenkalti/backoff/v4 v4.3.0 github.com/cert-manager/go-pkcs12 v0.0.0-20251218073410-44b982790b7c github.com/coreos/go-iptables v0.8.0 @@ -90,7 +92,8 @@ require ( golang.org/x/text v0.40.0 // indirect golang.org/x/time v0.15.0 // indirect golang.org/x/tools v0.47.0 // indirect - golang.zx2c4.com/wireguard v0.0.0-20230325221338-052af4a8072b // indirect + golang.zx2c4.com/wintun v0.0.0-20230126152724-0fa3db229ce2 // indirect + golang.zx2c4.com/wireguard v0.0.0-20250521234502-f333402bd9cb // indirect gopkg.in/evanphx/json-patch.v4 v4.13.0 // indirect gopkg.in/inf.v0 v0.9.1 // indirect gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect diff --git a/go.sum b/go.sum index da8039357..1120b9881 100644 --- a/go.sum +++ b/go.sum @@ -1,8 +1,12 @@ github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU= github.com/Masterminds/semver/v3 v3.4.0 h1:Zog+i5UMtVoCU8oKka5P7i9q9HgrJeGzI9SA1Xbatp0= github.com/Masterminds/semver/v3 v3.4.0/go.mod h1:4V+yj/TJE1HU9XfppCwVMZq3I84lprf4nC11bSS5beM= +github.com/advanced-wg/awgctrl-go v1.0.0 h1:Zkp90k7n9uWDni0OlDFdLPa7DHXNG0KJhkK8fxFNRH0= +github.com/advanced-wg/awgctrl-go v1.0.0/go.mod h1:uYzDR5QOkzOglbeoExOwUh9xEWUGleggeUEHCIR/1pY= github.com/alexflint/go-filemutex v0.0.0-20171022225611-72bdc8eae2ae h1:AMzIhMUqU3jMrZiTuW0zkYeKlKDAFD+DG20IoO421/Y= github.com/alexflint/go-filemutex v0.0.0-20171022225611-72bdc8eae2ae/go.mod h1:CgnQgUtFrFz9mxFNtED3jI5tLDjKlOM+oUF/sTk6ps0= +github.com/amnezia-vpn/amneziawg-go v0.2.20-0.20260709022233-c1e9bb3758e7 h1:OfKNgJiE0RnUv7gicBPe9bUm4ZEJDyPlb7MuLvQQHOg= +github.com/amnezia-vpn/amneziawg-go v0.2.20-0.20260709022233-c1e9bb3758e7/go.mod h1:5vpNSF+W7zANuHlrh5uX6IS2dsE7L6zipKaSRNSv20s= github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5 h1:0CwZNZbxp69SHPdPJAN/hZIm0C4OItdklCFmMRWYpio= github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5/go.mod h1:wHh0iHkYZB8zMSxRWpUBQtwG5a7fFgvEO+odwuTv2gs= github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= @@ -65,6 +69,8 @@ github.com/go-task/slim-sprig/v3 v3.0.0 h1:sUs3vkvUymDpBKi3qH1YSqBQk9+9D/8M2mN1v github.com/go-task/slim-sprig/v3 v3.0.0/go.mod h1:W848ghGpv3Qj3dhTPRyJypKRiqCdHZiAzKg9hl15HA8= github.com/goccy/go-yaml v1.18.0 h1:8W7wMFS12Pcas7KU+VVkaiCng+kG8QiFeFwzFb+rwuw= github.com/goccy/go-yaml v1.18.0/go.mod h1:XBurs7gK8ATbW4ZPGKgcbrY1Br56PdM69F7LkFRi1kA= +github.com/google/btree v1.1.3 h1:CVpQJjYgC4VbzxeGVHfvZrv1ctoYCAI8vbl07Fcxlyg= +github.com/google/btree v1.1.3/go.mod h1:qOPhT0dTNdNzV6Z/lhRX0YXUafgPLFUh+gZMl761Gm4= github.com/google/gnostic-models v0.7.0 h1:qwTtogB15McXDaNqTZdzPJRHvaVJlAl+HVQnLmJEJxo= github.com/google/gnostic-models v0.7.0/go.mod h1:whL5G0m6dmc5cPxKc5bdKdEN3UjI7OUGxBlw57miDrQ= github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= @@ -203,6 +209,8 @@ github.com/vishvananda/netns v0.0.5 h1:DfiHV+j8bA32MFM7bfEunvT8IAqQ/NzSJHtcmW5zd github.com/vishvananda/netns v0.0.5/go.mod h1:SpkAiCQRtJ6TvvxPnOSyH3BMl6unz3xZlaprSwhNNJM= github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM= github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg= +go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE= +go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0= go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0= @@ -236,8 +244,10 @@ golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno= golang.org/x/tools v0.47.0 h1:7Kn5x/d1svx/PzryTsqeoZN4TZwqeH5pGWjefhLi/1Q= golang.org/x/tools v0.47.0/go.mod h1:dFHnyTvFWY212G+h7ZY4Vsp/K3U4/7W9TyVaAul8uCA= -golang.zx2c4.com/wireguard v0.0.0-20230325221338-052af4a8072b h1:J1CaxgLerRR5lgx3wnr6L04cJFbWoceSK9JWBdglINo= -golang.zx2c4.com/wireguard v0.0.0-20230325221338-052af4a8072b/go.mod h1:tqur9LnfstdR9ep2LaJT4lFUl0EjlHtge+gAjmsHUG4= +golang.zx2c4.com/wintun v0.0.0-20230126152724-0fa3db229ce2 h1:B82qJJgjvYKsXS9jeunTOisW56dUokqW/FOteYJJ/yg= +golang.zx2c4.com/wintun v0.0.0-20230126152724-0fa3db229ce2/go.mod h1:deeaetjYA+DHMHg+sMSMI58GrEteJUUzzw7en6TJQcI= +golang.zx2c4.com/wireguard v0.0.0-20250521234502-f333402bd9cb h1:whnFRlWMcXI9d+ZbWg+4sHnLp52d5yiIPUxMBSt4X9A= +golang.zx2c4.com/wireguard v0.0.0-20250521234502-f333402bd9cb/go.mod h1:rpwXGsirqLqN2L0JDJQlwOboGHmptD5ZD6T2VmcqhTw= golang.zx2c4.com/wireguard/wgctrl v0.0.0-20230429144221-925a1e7659e6 h1:CawjfCvYQH2OU3/TnxLx97WDSUDRABfT18pCOYwc2GE= golang.zx2c4.com/wireguard/wgctrl v0.0.0-20230429144221-925a1e7659e6/go.mod h1:3rxYc4HtVcSG9gVaTs2GEBdehh+sYPOwKtyUWEOTb80= google.golang.org/protobuf v1.36.12-0.20260120151049-f2248ac996af h1:+5/Sw3GsDNlEmu7TfklWKPdQ0Ykja5VEmq2i817+jbI= @@ -263,6 +273,8 @@ gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gvisor.dev/gvisor v0.0.0-20250503011706-39ed1f5ac29c h1:m/r7OM+Y2Ty1sgBQ7Qb27VgIMBW8ZZhT4gLnUyDIhzI= +gvisor.dev/gvisor v0.0.0-20250503011706-39ed1f5ac29c/go.mod h1:3r5CMtNQMKIvBlrmM9xWUNamjKBYPOWyXOjmg5Kts3g= k8s.io/api v0.36.2 h1:TF6YDLIzKfccK7cq9YpTcGX8TJmEkHVRv78DM51fRYY= k8s.io/api v0.36.2/go.mod h1:F4LbMO4brjZYh7yFkXWhynSvtB7YauxV4c+HHkNRGNg= k8s.io/apiextensions-apiserver v0.36.0 h1:Wt7E8J+VBCbj4FjiBfDTK/neXDDjyJVJc7xfuOHImZ0= diff --git a/pkg/cable/amneziawg/amneziawg_suite_test.go b/pkg/cable/amneziawg/amneziawg_suite_test.go new file mode 100644 index 000000000..8df2fdd22 --- /dev/null +++ b/pkg/cable/amneziawg/amneziawg_suite_test.go @@ -0,0 +1,45 @@ +/* +SPDX-License-Identifier: Apache-2.0 + +Copyright Contributors to the Submariner project. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + +http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package amneziawg_test + +import ( + "flag" + "testing" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + "github.com/submariner-io/admiral/pkg/log/kzerolog" +) + +func init() { + kzerolog.AddFlags(nil) +} + +var _ = BeforeSuite(func() { + flags := flag.NewFlagSet("kzerolog", flag.ExitOnError) + kzerolog.AddFlags(flags) + _ = flags.Parse([]string{"-v=4"}) + + kzerolog.InitK8sLogging() +}) + +func TestAmneziaWG(t *testing.T) { + RegisterFailHandler(Fail) + RunSpecs(t, "AmneziaWG Suite") +} diff --git a/pkg/cable/amneziawg/amneziawg_test.go b/pkg/cable/amneziawg/amneziawg_test.go new file mode 100644 index 000000000..364ca2ed6 --- /dev/null +++ b/pkg/cable/amneziawg/amneziawg_test.go @@ -0,0 +1,761 @@ +/* +SPDX-License-Identifier: Apache-2.0 + +Copyright Contributors to the Submariner project. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + +http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package amneziawg_test + +import ( + "context" + "errors" + "net" + "os" + "slices" + "strconv" + "strings" + "time" + + "github.com/advanced-wg/awgctrl-go/wgtypes" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + "github.com/submariner-io/admiral/pkg/resource" + submarinerv1 "github.com/submariner-io/submariner/pkg/apis/submariner.io/v1" + "github.com/submariner-io/submariner/pkg/cable" + "github.com/submariner-io/submariner/pkg/cable/amneziawg" + "github.com/submariner-io/submariner/pkg/endpoint" + "github.com/submariner-io/submariner/pkg/natdiscovery" + netlinkAPI "github.com/submariner-io/submariner/pkg/netlink" + fakeNetlink "github.com/submariner-io/submariner/pkg/netlink/fake" + "github.com/submariner-io/submariner/pkg/types" + "github.com/vishvananda/netlink" + dynamicfake "k8s.io/client-go/dynamic/fake" + "k8s.io/client-go/kubernetes/scheme" + k8snet "k8s.io/utils/net" + "k8s.io/utils/ptr" +) + +const ( + listenPort = 123 + listenPort2 = 456 +) + +var _ = Describe("Driver", func() { + Context("NewDriver", testNewDriver) + Context("Init", testInit) + Context("ConnectToEndpoint", testConnectToEndpoint) + Context("DisconnectFromEndpoint", testDisconnectFromEndpoint) + Context("GetConnections", testGetConnections) + Context("Cleanup", testCleanup) + + Context("", func() { + t := newTestDriver() + + Specify("GetName should return amneziawg", func() { + Expect(t.driver.GetName()).To(Equal("amneziawg")) + }) + }) +}) + +type noopUserspaceDevice struct{} + +func (noopUserspaceDevice) Close() error { return nil } + +type testDriver struct { + endpointSpec submarinerv1.EndpointSpec + localEndpoint *endpoint.Local + driver cable.Driver + netLink *fakeNetlink.NetLink + client *fakeClient + checkNewDriverErr func(error) +} + +func newTestDriver() *testDriver { + t := &testDriver{} + + BeforeEach(func() { + os.Setenv("CE_IPSEC_PSK", "test-psk-value") + os.Unsetenv(cable.CableDriverOptionsEnv) + + t.endpointSpec = submarinerv1.EndpointSpec{ + ClusterID: "local", + CableName: "submariner-cable-local-192-68-1-1", + PrivateIPs: []string{"192.68.1.1"}, + Subnets: []string{"10.0.0.0/16"}, + BackendConfig: map[string]string{ + submarinerv1.UDPPortConfig: strconv.Itoa(listenPort), + }, + } + + t.netLink = fakeNetlink.New() + netlinkAPI.NewFunc = func() netlinkAPI.Interface { + return t.netLink + } + + t.client = &fakeClient{} + t.client.devices = map[string]*wgtypes.Device{} + + amneziawg.NewClient = func() (amneziawg.Client, error) { + return t.client, nil + } + + amneziawg.StartUserspaceDevice = func(ifaceName string) (amneziawg.UserspaceDevice, error) { + _ = t.netLink.LinkAdd(&netlink.GenericLink{ + LinkAttrs: netlink.LinkAttrs{ + Name: ifaceName, + }, + }) + + return noopUserspaceDevice{}, nil + } + + t.checkNewDriverErr = func(err error) { + Expect(err).NotTo(HaveOccurred()) + } + }) + + JustBeforeEach(func() { + var err error + + t.localEndpoint = endpoint.NewLocal(&t.endpointSpec, dynamicfake.NewSimpleDynamicClient(scheme.Scheme), "") + + t.driver, err = amneziawg.NewDriver(t.localEndpoint, &types.SubmarinerCluster{}, nil) + t.checkNewDriverErr(err) + }) + + return t +} + +func testNewDriver() { + t := newTestDriver() + + BeforeEach(func() { + t.client.devices[amneziawg.DefaultDeviceName] = &wgtypes.Device{ + Peers: []wgtypes.Peer{{PublicKey: wgtypes.Key{}}}, + } + }) + + It("should configure the device with obfuscation parameters and public key", func() { + device := t.client.devices[amneziawg.DefaultDeviceName] + Expect(device).ToNot(BeNil()) + Expect(device.ListenPort).To(Equal(listenPort)) + Expect(device.PublicKey).ToNot(Equal(wgtypes.Key{})) + Expect(device.PrivateKey).ToNot(Equal(wgtypes.Key{})) + Expect(device.Jc).To(Equal(7)) + Expect(device.Jmin).To(Equal(80)) + Expect(device.Jmax).To(Equal(200)) + Expect(device.S1).To(Equal(45)) + Expect(device.S2).To(Equal(60)) + Expect(device.S3).To(Equal(35)) + Expect(device.S4).To(Equal(12)) + Expect(device.H1).To(Equal("200000000-280000000")) + Expect(device.H2).To(Equal("400000000-480000000")) + Expect(device.I1).To(Equal("")) + Expect(device.Peers).To(BeEmpty(), "ReplacePeers should clear the pre-seeded peer") + + Expect(t.localEndpoint.Spec().BackendConfig[amneziawg.PublicKey]).To(Equal(device.PublicKey.String())) + Expect(t.localEndpoint.Spec().BackendConfig[cable.InterfaceNameConfig]).To(Equal(amneziawg.DefaultDeviceName)) + }) + + When("cable driver options are set", func() { + BeforeEach(func() { + os.Setenv(cable.CableDriverOptionsEnv, `{"jc":"3","h1":"10-20","i1":""}`) + }) + + AfterEach(func() { + os.Unsetenv(cable.CableDriverOptionsEnv) + }) + + It("should override the default obfuscation parameters", func() { + device := t.client.devices[amneziawg.DefaultDeviceName] + Expect(device.Jc).To(Equal(3)) + Expect(device.H1).To(Equal("10-20")) + Expect(device.I1).To(Equal("")) + Expect(device.S3).To(Equal(35)) + }) + }) + + When("cable driver options are invalid", func() { + BeforeEach(func() { + os.Setenv(cable.CableDriverOptionsEnv, `{"jc":"0"}`) + + t.checkNewDriverErr = func(err error) { + Expect(err).To(HaveOccurred()) + } + }) + + AfterEach(func() { + os.Unsetenv(cable.CableDriverOptionsEnv) + }) + + It("should return an error", func() { + }) + }) + + When("configuring the AmneziaWG device fails", func() { + BeforeEach(func() { + t.client.configureDeviceErr = errors.New("mock error") + t.checkNewDriverErr = func(err error) { + Expect(err).To(HaveOccurred()) + } + }) + + It("should return an error", func() { + }) + }) + + When("creating the AmneziaWG client fails", func() { + BeforeEach(func() { + amneziawg.NewClient = func() (amneziawg.Client, error) { + return nil, errors.New("mock") + } + + t.checkNewDriverErr = func(err error) { + Expect(err).To(HaveOccurred()) + } + }) + + It("should return an error", func() { + }) + }) + + When("the backend port is invalid", func() { + BeforeEach(func() { + t.endpointSpec.BackendConfig[submarinerv1.UDPPortConfig] = "bogus" + t.checkNewDriverErr = func(err error) { + Expect(err).To(HaveOccurred()) + } + }) + + It("should return an error", func() { + }) + }) +} + +func testInit() { + t := newTestDriver() + + It("should succeed", func() { + Expect(t.driver.Init(context.TODO())).To(Succeed()) + t.netLink.AwaitLinkSetup(amneziawg.DefaultDeviceName) + }) + + When("link setup fails", func() { + It("should return an error", func() { + link := t.netLink.AwaitLink(amneziawg.DefaultDeviceName) + _ = t.netLink.LinkDel(link) + + Expect(t.driver.Init(context.TODO())).NotTo(Succeed()) + }) + }) +} + +func testConnectToEndpoint() { + t := newTestDriver() + + When("keepalive cable driver option is set", func() { + BeforeEach(func() { + os.Setenv(cable.CableDriverOptionsEnv, `{"keepalive":"25"}`) + }) + + AfterEach(func() { + os.Unsetenv(cable.CableDriverOptionsEnv) + }) + + It("should use the configured keepalive when connecting", func() { + natInfo := newNATInfo("remote", "172.16.0.0/16") + _, err := t.driver.ConnectToEndpoint(natInfo) + Expect(err).NotTo(HaveOccurred()) + + device, err := t.client.Device(amneziawg.DefaultDeviceName) + Expect(err).NotTo(HaveOccurred()) + Expect(device.Peers).To(HaveLen(1)) + Expect(device.Peers[0].PersistentKeepaliveInterval).To(Equal(25 * time.Second)) + }) + }) + + It("should create a Connection and configure a peer on the AmneziaWG device", func() { + natInfo := newNATInfo("east", "20.0.0.0/16", "30.0.0.0/16") + + ip, err := t.driver.ConnectToEndpoint(natInfo) + Expect(err).To(Succeed()) + Expect(ip).To(Equal(natInfo.UseIP)) + + t.client.assertDevicePeers(natInfo) + t.assertConnections(natInfo) + + // Calling ConnectToEndpoint again with the same endpoint should essentially be a no-op. + + ip, err = t.driver.ConnectToEndpoint(natInfo) + Expect(err).To(Succeed()) + Expect(ip).To(Equal(natInfo.UseIP)) + + t.client.assertDevicePeers(natInfo) + t.assertConnections(natInfo) + + // Calling ConnectToEndpoint again with a differing endpoint from the same cluster should replace the previous. + + priv, err := wgtypes.GeneratePrivateKey() + Expect(err).To(Succeed()) + + natInfo.Endpoint = *natInfo.Endpoint.DeepCopy() + natInfo.Endpoint.Spec.BackendConfig[amneziawg.PublicKey] = priv.PublicKey().String() + natInfo.Endpoint.Spec.Subnets = []string{"40.0.0.0/16"} + + ip, err = t.driver.ConnectToEndpoint(natInfo) + Expect(err).ToNot(HaveOccurred()) + Expect(ip).To(Equal(natInfo.UseIP)) + + t.client.assertDevicePeers(natInfo) + t.assertConnections(natInfo) + + // Connect to an endpoint from a different cluster + + natInfo2 := newNATInfo("west", "50.0.0.0/16") + + ip, err = t.driver.ConnectToEndpoint(natInfo2) + Expect(err).ToNot(HaveOccurred()) + Expect(ip).To(Equal(natInfo2.UseIP)) + + t.client.assertDevicePeers(natInfo, natInfo2) + t.assertConnections(natInfo, natInfo2) + + Expect(t.driver.GetActiveConnections()).To(BeEmpty()) + }) + + When("configuring the AmneziaWG device fails", func() { + It("should return an error", func() { + t.client.configureDeviceErr = errors.New("mock error") + + _, err := t.driver.ConnectToEndpoint(newNATInfo("east")) + Expect(err).To(HaveOccurred()) + }) + }) + + When("the public key is missing from the remote endpoint", func() { + It("should return an error", func() { + natInfo := newNATInfo("east") + natInfo.Endpoint.Spec.BackendConfig = map[string]string{} + + _, err := t.driver.ConnectToEndpoint(natInfo) + Expect(err).To(HaveOccurred()) + }) + }) +} + +func testDisconnectFromEndpoint() { + t := newTestDriver() + + It("should remove the Connection and the AmneziaWG device peer", func() { + natInfo := newNATInfo("east", "20.0.0.0/16") + + _, err := t.driver.ConnectToEndpoint(natInfo) + Expect(err).ToNot(HaveOccurred()) + + natInfo2 := newNATInfo("west", "21.0.0.0/16") + + _, err = t.driver.ConnectToEndpoint(natInfo2) + Expect(err).ToNot(HaveOccurred()) + + err = t.driver.DisconnectFromEndpoint(&types.SubmarinerEndpoint{Spec: natInfo.Endpoint.Spec}, k8snet.IPv4) + Expect(err).ToNot(HaveOccurred()) + + t.client.assertDevicePeers(natInfo2) + t.assertConnections(natInfo2) + + err = t.driver.DisconnectFromEndpoint(&types.SubmarinerEndpoint{Spec: natInfo2.Endpoint.Spec}, k8snet.IPv4) + Expect(err).ToNot(HaveOccurred()) + + t.client.assertDevicePeers() + t.assertConnections() + }) + + When("the public key is missing from the remote endpoint", func() { + It("should return an error", func() { + natInfo := newNATInfo("east") + natInfo.Endpoint.Spec.BackendConfig = map[string]string{} + + err := t.driver.DisconnectFromEndpoint(&types.SubmarinerEndpoint{Spec: natInfo.Endpoint.Spec}, k8snet.IPv4) + Expect(err).To(HaveOccurred()) + }) + }) + + When("the remote endpoint public key does not match that of the prior connection from the same cluster", func() { + It("should return not remove the Connection", func() { + err := t.driver.DisconnectFromEndpoint(&types.SubmarinerEndpoint{Spec: newNATInfo("east").Endpoint.Spec}, k8snet.IPv4) + Expect(err).ToNot(HaveOccurred()) + + natInfo := newNATInfo("east", "20.0.0.0/16") + + _, err = t.driver.ConnectToEndpoint(natInfo) + Expect(err).ToNot(HaveOccurred()) + + err = t.driver.DisconnectFromEndpoint(&types.SubmarinerEndpoint{Spec: newNATInfo("east").Endpoint.Spec}, k8snet.IPv4) + Expect(err).ToNot(HaveOccurred()) + + t.assertConnections(natInfo) + }) + }) +} + +func testGetConnections() { + t := newTestDriver() + + BeforeEach(func() { + // Disable peer keepalive so the no-traffic grace window also uses StatusPollInterval + // (otherwise default keepalive=10s would delay soft ConnectionError in this suite). + os.Setenv(cable.CableDriverOptionsEnv, `{"keepalive":"0"}`) + + oldStatusPollInterval := amneziawg.StatusPollInterval + amneziawg.StatusPollInterval = time.Millisecond * 50 + + oldHandshakeTimeout := amneziawg.HandshakeTimeout + amneziawg.HandshakeTimeout = time.Millisecond * 100 + + DeferCleanup(func() { + os.Unsetenv(cable.CableDriverOptionsEnv) + + amneziawg.StatusPollInterval = oldStatusPollInterval + amneziawg.HandshakeTimeout = oldHandshakeTimeout + }) + }) + + getConnection := func() *submarinerv1.Connection { + conns, err := t.driver.GetConnections() + Expect(err).ToNot(HaveOccurred()) + Expect(conns).To(HaveLen(1)) + + return &conns[0] + } + + It("should correctly update the peer connection status", func() { + _, err := t.driver.ConnectToEndpoint(newNATInfo("east", "20.0.0.0/16")) + Expect(err).To(Succeed()) + + Expect(t.client.devices[amneziawg.DefaultDeviceName].Peers).To(HaveLen(1)) + peer := &t.client.devices[amneziawg.DefaultDeviceName].Peers[0] + + _ = getConnection() + + By("No change - should remain Connecting") + + time.Sleep(amneziawg.StatusPollInterval + time.Millisecond*5) + + conn := getConnection() + Expect(conn.Status).To(Equal(submarinerv1.Connecting), "Unexpected status %q", conn.StatusMessage) + + By("Initial handshake timeout - should report ConnectionError") + + time.Sleep(amneziawg.HandshakeTimeout + time.Millisecond*10) + + conn = getConnection() + Expect(conn.Status).To(Equal(submarinerv1.ConnectionError), "Unexpected status %q", conn.StatusMessage) + + By("Clear handshake timeout and add Tx bytes - should go back to Connecting since no handshake yet") + + time.Sleep(amneziawg.StatusPollInterval + time.Millisecond*5) + + amneziawg.HandshakeTimeout += time.Minute + peer.TransmitBytes = 1000 + conn = getConnection() + + Expect(conn.Status).To(Equal(submarinerv1.Connecting), "Unexpected status %q", conn.StatusMessage) + + By("Set that handshake occurred and add Rx bytes - should report Connected") + + peer.LastHandshakeTime = time.Now() + peer.ReceiveBytes += 1000 + + time.Sleep(amneziawg.StatusPollInterval + time.Millisecond*5) + + conn = getConnection() + Expect(conn.Status).To(Equal(submarinerv1.Connected), "Unexpected status %q", conn.StatusMessage) + + By("No change - should remain Connected") + + time.Sleep(amneziawg.StatusPollInterval + time.Millisecond*5) + + conn = getConnection() + Expect(conn.Status).To(Equal(submarinerv1.Connected), "Unexpected status %q", conn.StatusMessage) + + By("No traffic - handshake stale - should report ConnectionError") + + time.Sleep(amneziawg.StatusPollInterval + time.Millisecond*5) + + conn = getConnection() + Expect(conn.Status).To(Equal(submarinerv1.ConnectionError), "Unexpected status %q", conn.StatusMessage) + + By("Add Tx/Rx bytes - should report Connected") + + peer.ReceiveBytes += 1000 + peer.TransmitBytes += 1000 + + time.Sleep(amneziawg.StatusPollInterval + time.Millisecond*5) + + conn = getConnection() + Expect(conn.Status).To(Equal(submarinerv1.Connected), "Unexpected status %q", conn.StatusMessage) + + By("No traffic and handshake timeout - should report ConnectionError") + + time.Sleep(amneziawg.StatusPollInterval + time.Millisecond*5) + + peer.LastHandshakeTime = time.Now().Add(-amneziawg.HandshakeTimeout) + conn = getConnection() + Expect(conn.Status).To(Equal(submarinerv1.ConnectionError), "Unexpected status %q", conn.StatusMessage) + + By("Clear handshake timeout and add Tx/Rx bytes - should report Connected") + + peer.LastHandshakeTime = time.Now() + peer.ReceiveBytes += 1000 + peer.TransmitBytes += 1000 + + time.Sleep(amneziawg.StatusPollInterval + time.Millisecond*5) + + conn = getConnection() + Expect(conn.Status).To(Equal(submarinerv1.Connected), "Unexpected status %q", conn.StatusMessage) + }) + + Context("with a stale peer present", func() { + It("should remove the stale peer", func() { + key, err := wgtypes.GenerateKey() + Expect(err).ToNot(HaveOccurred()) + + t.client.devices[amneziawg.DefaultDeviceName].Peers = append(t.client.devices[amneziawg.DefaultDeviceName].Peers, + wgtypes.Peer{PublicKey: key}) + + _, err = t.driver.GetConnections() + Expect(err).ToNot(HaveOccurred()) + + Expect(t.client.devices[amneziawg.DefaultDeviceName].Peers).To(BeEmpty()) + }) + }) + + When("device retrieval fails", func() { + It("should return an error", func() { + t.client.devices = map[string]*wgtypes.Device{} + _, err := t.driver.GetConnections() + Expect(err).To(HaveOccurred()) + }) + }) +} + +func testCleanup() { + t := newTestDriver() + + It("should delete the device link", func() { + Expect(t.driver.Cleanup(context.TODO())).To(Succeed()) + t.netLink.AwaitNoLink(amneziawg.DefaultDeviceName) + }) +} + +func (t *testDriver) assertConnections(natInfos ...*natdiscovery.NATEndpointInfo) { + actual, err := t.driver.GetConnections() + Expect(err).ToNot(HaveOccurred()) + + for i := range actual { + actual[i].StatusMessage = "" + } + + slices.SortFunc(actual, func(a, b submarinerv1.Connection) int { + return strings.Compare(a.Endpoint.BackendConfig[amneziawg.PublicKey], b.Endpoint.BackendConfig[amneziawg.PublicKey]) + }) + + expected := make([]submarinerv1.Connection, len(natInfos)) + for i := range natInfos { + expected[i] = submarinerv1.Connection{ + Status: submarinerv1.Connecting, + Endpoint: natInfos[i].Endpoint.Spec, + UsingIP: natInfos[i].UseIP, + UsingNAT: natInfos[i].UseNAT, + } + } + + slices.SortFunc(expected, func(a, b submarinerv1.Connection) int { + return strings.Compare(a.Endpoint.BackendConfig[amneziawg.PublicKey], b.Endpoint.BackendConfig[amneziawg.PublicKey]) + }) + + Expect(actual).To(HaveExactElements(expected)) +} + +func newNATInfo(clusterID string, subnets ...string) *natdiscovery.NATEndpointInfo { + priv, err := wgtypes.GeneratePrivateKey() + Expect(err).ToNot(HaveOccurred()) + + return &natdiscovery.NATEndpointInfo{ + Endpoint: submarinerv1.Endpoint{ + Spec: submarinerv1.EndpointSpec{ + ClusterID: clusterID, + CableName: "submariner-cable-" + clusterID, + Subnets: subnets, + BackendConfig: map[string]string{ + amneziawg.PublicKey: priv.PublicKey().String(), + submarinerv1.UDPPortConfig: strconv.Itoa(listenPort2), + }, + }, + }, + UseIP: "172.93.2.1", + UseNAT: true, + UseFamily: k8snet.IPv4, + } +} + +type fakeClient struct { + devices map[string]*wgtypes.Device + configureDeviceErr error +} + +//nolint:gocritic // hugeParam: matches amneziawg.Client / awgctrl API +func (c *fakeClient) ConfigureDevice(name string, cfg wgtypes.Config) error { + if c.configureDeviceErr != nil { + return c.configureDeviceErr + } + + d := c.devices[name] + if d == nil { + c.devices[name] = &wgtypes.Device{} + d = c.devices[name] + } + + if cfg.PrivateKey != nil { + d.PrivateKey = *cfg.PrivateKey + d.PublicKey = d.PrivateKey.PublicKey() + } + + if cfg.ListenPort != nil { + d.ListenPort = *cfg.ListenPort + } + + applyIntOpt := func(dst *int, src *int) { + if src != nil { + *dst = *src + } + } + + applyStrOpt := func(dst *string, src *string) { + if src != nil { + *dst = *src + } + } + + applyIntOpt(&d.Jc, cfg.Jc) + applyIntOpt(&d.Jmin, cfg.Jmin) + applyIntOpt(&d.Jmax, cfg.Jmax) + applyIntOpt(&d.S1, cfg.S1) + applyIntOpt(&d.S2, cfg.S2) + applyIntOpt(&d.S3, cfg.S3) + applyIntOpt(&d.S4, cfg.S4) + applyStrOpt(&d.H1, cfg.H1) + applyStrOpt(&d.H2, cfg.H2) + applyStrOpt(&d.H3, cfg.H3) + applyStrOpt(&d.H4, cfg.H4) + applyStrOpt(&d.I1, cfg.I1) + applyStrOpt(&d.I2, cfg.I2) + applyStrOpt(&d.I3, cfg.I3) + applyStrOpt(&d.I4, cfg.I4) + applyStrOpt(&d.I5, cfg.I5) + + if cfg.ReplacePeers { + d.Peers = nil + } + + for i := range cfg.Peers { + pc := &cfg.Peers[i] + if pc.Remove { + d.Peers = slices.DeleteFunc(d.Peers, func(p wgtypes.Peer) bool { + return p.PublicKey.String() == pc.PublicKey.String() + }) + + continue + } + + index := slices.IndexFunc(d.Peers, func(p wgtypes.Peer) bool { + return p.PublicKey.String() == pc.PublicKey.String() + }) + + if index == -1 { + if pc.UpdateOnly { + continue + } + + d.Peers = append(d.Peers, wgtypes.Peer{ + PublicKey: pc.PublicKey, + PresharedKey: ptr.Deref(pc.PresharedKey, wgtypes.Key{}), + Endpoint: pc.Endpoint, + PersistentKeepaliveInterval: ptr.Deref(pc.PersistentKeepaliveInterval, 0), + }) + + index = len(d.Peers) - 1 + } + + peer := &d.Peers[index] + + if pc.ReplaceAllowedIPs { + peer.AllowedIPs = pc.AllowedIPs + } else { + peer.AllowedIPs = append(peer.AllowedIPs, pc.AllowedIPs...) + } + } + + return nil +} + +func (c *fakeClient) Device(name string) (*wgtypes.Device, error) { + if c.devices[name] != nil { + d := *c.devices[name] + d.Peers = make([]wgtypes.Peer, len(c.devices[name].Peers)) + copy(d.Peers, c.devices[name].Peers) + + return &d, nil + } + + return nil, os.ErrNotExist +} + +func (c *fakeClient) Close() error { + return nil +} + +func (c *fakeClient) assertDevicePeers(natInfos ...*natdiscovery.NATEndpointInfo) { + device, err := c.Device(amneziawg.DefaultDeviceName) + Expect(err).ToNot(HaveOccurred()) + + for i := range natInfos { + index := slices.IndexFunc(device.Peers, func(p wgtypes.Peer) bool { + return p.PublicKey.String() == natInfos[i].Endpoint.Spec.BackendConfig[amneziawg.PublicKey] + }) + Expect(index).To(BeNumerically(">=", 0), "Missing expected device peer for %s", resource.ToJSON(natInfos[i])) + + peer := &device.Peers[index] + Expect(peer.PublicKey.String()).To(Equal(natInfos[i].Endpoint.Spec.BackendConfig[amneziawg.PublicKey])) + Expect(peer.PresharedKey).ToNot(Equal(wgtypes.Key{})) + Expect(peer.Endpoint).To(Equal(&net.UDPAddr{ + IP: net.ParseIP(natInfos[i].UseIP), + Port: listenPort2, + })) + + actualIPs := make([]string, len(peer.AllowedIPs)) + for j := range peer.AllowedIPs { + actualIPs[j] = peer.AllowedIPs[j].String() + } + + slices.Sort(actualIPs) + slices.Sort(natInfos[i].Endpoint.Spec.Subnets) + Expect(actualIPs).To(HaveExactElements(natInfos[i].Endpoint.Spec.Subnets)) + + device.Peers = slices.Delete(device.Peers, index, index+1) + } + + Expect(device.Peers).To(BeEmpty(), "Received unexpected device peers") +} diff --git a/pkg/cable/amneziawg/client.go b/pkg/cable/amneziawg/client.go new file mode 100644 index 000000000..f186b893c --- /dev/null +++ b/pkg/cable/amneziawg/client.go @@ -0,0 +1,59 @@ +/* +SPDX-License-Identifier: Apache-2.0 + +Copyright Contributors to the Submariner project. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + +http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package amneziawg + +import ( + "context" + + awgctrl "github.com/advanced-wg/awgctrl-go" + "github.com/advanced-wg/awgctrl-go/wgtypes" + "github.com/pkg/errors" +) + +type Client interface { + ConfigureDevice(name string, cfg wgtypes.Config) error + Device(name string) (*wgtypes.Device, error) + Close() error +} + +type awgClient struct { + client *awgctrl.Client +} + +//nolint:gocritic // hugeParam: matches awgctrl.Client API +func (c *awgClient) ConfigureDevice(name string, cfg wgtypes.Config) error { + return c.client.ConfigureDevice(context.Background(), name, cfg) //nolint:wrapcheck // Let the caller wrap it +} + +func (c *awgClient) Device(name string) (*wgtypes.Device, error) { + return c.client.Device(context.Background(), name) //nolint:wrapcheck // Let the caller wrap it +} + +func (c *awgClient) Close() error { + return c.client.Close() //nolint:wrapcheck // Let the caller wrap it +} + +var NewClient = func() (Client, error) { + client, err := awgctrl.New() + if err != nil { + return nil, errors.Wrap(err, "failed to create awgctrl client") + } + + return &awgClient{client: client}, nil +} diff --git a/pkg/cable/amneziawg/daemon.go b/pkg/cable/amneziawg/daemon.go new file mode 100644 index 000000000..dcb751916 --- /dev/null +++ b/pkg/cable/amneziawg/daemon.go @@ -0,0 +1,113 @@ +/* +SPDX-License-Identifier: Apache-2.0 + +Copyright Contributors to the Submariner project. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + +http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package amneziawg + +import ( + "fmt" + "net" + "sync" + + "github.com/amnezia-vpn/amneziawg-go/conn" + "github.com/amnezia-vpn/amneziawg-go/device" + "github.com/amnezia-vpn/amneziawg-go/ipc" + "github.com/amnezia-vpn/amneziawg-go/tun" + "github.com/pkg/errors" +) + +// UserspaceDevice is the embedded amneziawg-go daemon. Unit tests may substitute a no-op. +type UserspaceDevice interface { + Close() error +} + +type embeddedDaemon struct { + tunDev tun.Device + dev *device.Device + uapi net.Listener + wg sync.WaitGroup + once sync.Once +} + +func (d *embeddedDaemon) Close() error { + var err error + + d.once.Do(func() { + if d.uapi != nil { + err = d.uapi.Close() + } + + if d.dev != nil { + d.dev.Close() + } + + d.wg.Wait() + }) + + return errors.Wrap(err, "error closing AmneziaWG UAPI listener") +} + +// StartUserspaceDevice creates the embedded amneziawg-go daemon. Unit tests may override this. +var StartUserspaceDevice = func(ifaceName string) (UserspaceDevice, error) { + tdev, err := tun.CreateTUN(ifaceName, device.DefaultMTU) + if err != nil { + return nil, errors.Wrap(err, "failed to create TUN device") + } + + fileUAPI, err := ipc.UAPIOpen(ifaceName) + if err != nil { + _ = tdev.Close() + return nil, errors.Wrap(err, "failed to open UAPI socket") + } + + devLogger := device.NewLogger(device.LogLevelError, fmt.Sprintf("(%s) ", ifaceName)) + dev := device.NewDevice(tdev, conn.NewDefaultBind(), devLogger) + + uapi, err := ipc.UAPIListen(ifaceName, fileUAPI) + if err != nil { + _ = fileUAPI.Close() + _ = tdev.Close() + dev.Close() + + return nil, errors.Wrap(err, "failed to listen on UAPI socket") + } + + daemon := &embeddedDaemon{ + tunDev: tdev, + dev: dev, + uapi: uapi, + } + + daemon.wg.Go(func() { + for { + conn, acceptErr := uapi.Accept() + if acceptErr != nil { + if errors.Is(acceptErr, net.ErrClosed) { + logger.Info("UAPI listener closed") + } else { + logger.Errorf(acceptErr, "UAPI accept error, stopping accept loop") + } + + return + } + + go dev.IpcHandle(conn) + } + }) + + return daemon, nil +} diff --git a/pkg/cable/amneziawg/driver.go b/pkg/cable/amneziawg/driver.go new file mode 100644 index 000000000..8d2efacc0 --- /dev/null +++ b/pkg/cable/amneziawg/driver.go @@ -0,0 +1,495 @@ +/* +SPDX-License-Identifier: Apache-2.0 + +Copyright Contributors to the Submariner project. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + +http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package amneziawg + +import ( + "context" + "crypto/sha256" + "fmt" + "net" + "slices" + "time" + + "github.com/advanced-wg/awgctrl-go/wgtypes" + "github.com/kelseyhightower/envconfig" + "github.com/pkg/errors" + "github.com/submariner-io/admiral/pkg/certificate" + "github.com/submariner-io/admiral/pkg/log" + "github.com/submariner-io/admiral/pkg/resource" + v1 "github.com/submariner-io/submariner/pkg/apis/submariner.io/v1" + "github.com/submariner-io/submariner/pkg/cable" + "github.com/submariner-io/submariner/pkg/endpoint" + "github.com/submariner-io/submariner/pkg/natdiscovery" + netlinkAPI "github.com/submariner-io/submariner/pkg/netlink" + "github.com/submariner-io/submariner/pkg/types" + "github.com/vishvananda/netlink" + k8snet "k8s.io/utils/net" + logf "sigs.k8s.io/controller-runtime/pkg/log" +) + +const ( + // DefaultDeviceName specifies name of the AmneziaWG network device. + DefaultDeviceName = "submariner" + + // PublicKey is name (key) of publicKey entry in back-end map. + PublicKey = "publicKey" + + cableDriverName = "amneziawg" + receiveBytes = "ReceiveBytes" // for peer connection status + transmitBytes = "TransmitBytes" // for peer connection status + lastChecked = "LastChecked" // for connection peer status +) + +var ( + // StatusPollInterval is how often GetConnections re-evaluates peer status. + StatusPollInterval = 10 * time.Second + + // HandshakeTimeout is maximal time from handshake a connection is still considered connected. + HandshakeTimeout = 2*time.Minute + 10*time.Second +) + +var logger = log.Logger{Logger: logf.Log.WithName("amneziawg")} + +func init() { + cable.AddDriver(cableDriverName, NewDriver) +} + +type specification struct { + PSK string `default:"default psk"` + NATTPort int32 `default:"4500"` +} + +type amneziawgDriver struct { + localEndpoint v1.EndpointSpec + connections map[string]*v1.Connection // clusterID -> remote ep connection + client Client + netLink netlinkAPI.Interface + link netlink.Link + spec *specification + psk *wgtypes.Key + keepAlive time.Duration + userspaceDevice UserspaceDevice +} + +// NewDriver creates a new AmneziaWG cable driver. +func NewDriver(localEndpoint *endpoint.Local, _ *types.SubmarinerCluster, _ certificate.SigningRequestor) (cable.Driver, error) { + // We'll panic if localEndpoint is nil, this is intentional + var err error + + a := amneziawgDriver{ + connections: make(map[string]*v1.Connection), + spec: new(specification), + netLink: netlinkAPI.New(), + } + + if err = envconfig.Process(cable.IPSecEnvPrefix, a.spec); err != nil { + return nil, errors.Wrap(err, "error processing environment config for amneziawg") + } + + if a.spec.PSK == "" || a.spec.PSK == "default psk" { + return nil, errors.New("CE_IPSEC_PSK must be set to a strong random value for the AmneziaWG driver") + } + + // Track success explicitly so cleanup does not depend on which err was last assigned + // (short declarations with := can otherwise skip teardown on failure paths). + setupOK := false + + defer func() { + if setupOK { + return + } + + if a.client != nil { + if e := a.client.Close(); e != nil { + logger.Errorf(e, "Failed to close client") + } + + a.client = nil + } + + a.cleanupDevice() //nolint:errcheck // Best-effort cleanup on NewDriver failure + }() + + if err = a.setupDevice(); err != nil { + return nil, errors.Wrap(err, "failed to setup AmneziaWG device") + } + + if a.client, err = NewClient(); err != nil { + return nil, errors.Wrap(err, "failed to open awgctrl client") + } + + var priv, pub, psk wgtypes.Key + + if psk, err = genPsk(a.spec.PSK); err != nil { + return nil, errors.Wrap(err, "error generating pre-shared key") + } + + a.psk = &psk + + if priv, err = wgtypes.GeneratePrivateKey(); err != nil { + return nil, errors.Wrap(err, "error generating private key") + } + + var port int32 + + port, err = localEndpoint.Spec().GetBackendPort(v1.UDPPortConfig, a.spec.NATTPort) + if err != nil { + return nil, errors.Wrapf(err, "error parsing %q from local endpoint", v1.UDPPortConfig) + } + + cfg := wgtypes.Config{ + PrivateKey: &priv, + ListenPort: new(int(port)), + FirewallMark: nil, + ReplacePeers: true, + Peers: []wgtypes.PeerConfig{}, + } + + if err = applyCableDriverOptions(&a, &cfg); err != nil { + return nil, errors.Wrap(err, "error applying AmneziaWG cable driver options") + } + + if err = a.client.ConfigureDevice(DefaultDeviceName, cfg); err != nil { + return nil, errors.Wrap(err, "failed to configure AmneziaWG device") + } + + pub = priv.PublicKey() + + err = localEndpoint.Update(context.TODO(), func(existing *v1.EndpointSpec) { + existing.BackendConfig[PublicKey] = pub.String() + existing.BackendConfig[cable.InterfaceNameConfig] = DefaultDeviceName + }) + if err != nil { + return nil, errors.Wrap(err, "error updating local endpoint") + } + + a.localEndpoint = *localEndpoint.Spec() + + logger.V(log.DEBUG).Infof("Created AmneziaWG %s with publicKey %s", DefaultDeviceName, pub) + + setupOK = true + + return &a, nil +} + +func (a *amneziawgDriver) Init(_ context.Context) error { + logger.V(log.DEBUG).Infof("Initializing AmneziaWG device for cluster %s", a.localEndpoint.ClusterID) + + l, err := a.netLink.InterfaceByName(DefaultDeviceName) + if err != nil { + return errors.Wrapf(err, "cannot get AmneziaWG link by name %s", DefaultDeviceName) + } + + d, err := a.client.Device(DefaultDeviceName) + if err != nil { + return errors.Wrap(err, "awgctrl cannot find AmneziaWG device") + } + + k, _ := keyFromSpec(&a.localEndpoint) + if k.String() != d.PublicKey.String() { + return fmt.Errorf("endpoint public key %s is different from device key %s", k, d.PublicKey) + } + + if err := a.netLink.LinkSetUp(a.link); err != nil { + return errors.Wrap(err, "failed to bring up AmneziaWG device") + } + + logger.V(log.DEBUG).Infof("AmneziaWG device %s is up on i/f number %d, listening on port :%d, with key %s", + a.link.Attrs().Name, l.Index(), d.ListenPort, d.PublicKey) + + return nil +} + +func (a *amneziawgDriver) GetName() string { + return cableDriverName +} + +func (a *amneziawgDriver) ConnectToEndpoint(endpointInfo *natdiscovery.NATEndpointInfo) (string, error) { + // We'll panic if endpointInfo is nil, this is intentional + remoteEndpoint := &endpointInfo.Endpoint + ip := endpointInfo.UseIP + + remoteIP := net.ParseIP(ip) + if remoteIP == nil { + return "", fmt.Errorf("failed to parse remote IP %s", ip) + } + + allowedIPs := remoteEndpoint.Spec.ParseSubnets(endpointInfo.UseFamily) + + remoteKey, err := keyFromSpec(&remoteEndpoint.Spec) + if err != nil { + return "", errors.Wrapf(err, "failed to obtain public key for endpoint %s", resource.ToJSON(remoteEndpoint.Spec)) + } + + logger.V(log.DEBUG).Infof("Connecting cluster %q endpoint %q with publicKey %q", + remoteEndpoint.Spec.ClusterID, remoteIP, remoteKey) + + oldCon, found := a.connections[remoteEndpoint.Spec.ClusterID] + if found { + if oldKey, err := keyFromSpec(&oldCon.Endpoint); err == nil { + if oldKey.String() == remoteKey.String() { + a.updatePeerStatus(oldCon, oldKey) + logger.V(log.DEBUG).Infof("Skipping connect for existing peer key %q", oldKey) + + return ip, nil + } + + if err := a.removePeer(oldKey); err != nil { + logger.Warningf("Failed to remove old peer %q for cluster %q: %v", + oldKey, remoteEndpoint.Spec.ClusterID, err) + } + } + + delete(a.connections, remoteEndpoint.Spec.ClusterID) + } + + connection := v1.NewConnection(&remoteEndpoint.Spec, ip, endpointInfo.UseNAT) + connection.SetStatus(v1.Connecting, "Connection has been created but not yet started") + + port, err := remoteEndpoint.Spec.GetBackendPort(v1.UDPPortConfig, a.spec.NATTPort) + if err != nil { + logger.Warningf("Error parsing %q from remote endpoint %q - using port %d instead: %v", v1.UDPPortConfig, + remoteEndpoint.Spec.CableName, a.spec.NATTPort, err) + } + + remotePort := int(port) + + peerCfg := []wgtypes.PeerConfig{{ + PublicKey: *remoteKey, + Remove: false, + UpdateOnly: false, + PresharedKey: a.psk, + Endpoint: &net.UDPAddr{ + IP: remoteIP, + Port: remotePort, + }, + PersistentKeepaliveInterval: new(a.keepAlive), + ReplaceAllowedIPs: true, + AllowedIPs: allowedIPs, + }} + + err = a.client.ConfigureDevice(DefaultDeviceName, wgtypes.Config{ + ReplacePeers: false, + Peers: peerCfg, + }) + if err != nil { + return "", errors.Wrap(err, "failed to configure peer") + } + + err = a.verifyNewPeer(&peerCfg[0]) + if err != nil { + if remErr := a.removePeer(remoteKey); remErr != nil { + logger.Warningf("Failed to remove unverified peer %q: %v", remoteKey, remErr) + } + + return "", errors.Wrap(err, "failed to verify peer configuration") + } + + // Insert only after ConfigureDevice/verify succeed so failed attempts leave no stale map entry. + a.connections[remoteEndpoint.Spec.ClusterID] = connection + + logger.V(log.DEBUG).Infof("Added connection for cluster %q: %s", remoteEndpoint.Spec.ClusterID, + resource.ToJSON(connection)) + logger.V(log.DEBUG).Infof("Successfully connected endpoint peer %q with IP %q", *remoteKey, remoteIP) + + cable.RecordConnection(cableDriverName, &a.localEndpoint, &connection.Endpoint, string(v1.Connected), true, endpointInfo.UseFamily) + + return ip, nil +} + +func keyFromSpec(ep *v1.EndpointSpec) (*wgtypes.Key, error) { + s, found := ep.BackendConfig[PublicKey] + if !found { + return &wgtypes.Key{}, errors.New("endpoint is missing public key") + } + + key, err := wgtypes.ParseKey(s) + + return &key, errors.Wrapf(err, "failed to parse public key %s", s) +} + +func (a *amneziawgDriver) DisconnectFromEndpoint(remoteEndpoint *types.SubmarinerEndpoint, family k8snet.IPFamily) error { + // We'll panic if remoteEndpoint is nil, this is intentional + logger.V(log.DEBUG).Infof("Removing IPv%v endpoint %s", family, resource.ToJSON(remoteEndpoint)) + + remoteKey, err := keyFromSpec(&remoteEndpoint.Spec) + if err != nil { + return errors.Wrap(err, "failed to parse peer public key") + } + + if err := a.removePeer(remoteKey); err != nil { + logger.Warningf("Failed to remove peer %q for cluster %q: %v", + remoteKey, remoteEndpoint.Spec.ClusterID, err) + } + + if a.keyMismatch(remoteEndpoint.Spec.ClusterID, remoteKey) { + logger.Warningf("Key mismatch for peer cluster %s, keeping existing spec", remoteEndpoint.Spec.ClusterID) + return nil + } + + delete(a.connections, remoteEndpoint.Spec.ClusterID) + + logger.V(log.DEBUG).Infof("Done removing endpoint for cluster %q", remoteEndpoint.Spec.ClusterID) + cable.RecordDisconnected(cableDriverName, &a.localEndpoint, &remoteEndpoint.Spec, family) + + return nil +} + +func (a *amneziawgDriver) GetActiveConnections() ([]v1.Connection, error) { + // force caller to skip duplicate handling + return make([]v1.Connection, 0), nil +} + +func (a *amneziawgDriver) setupDevice() error { + link, err := a.netLink.LinkByName(DefaultDeviceName) + switch { + case err == nil: + if err := a.netLink.LinkDel(link); err != nil { + return errors.Wrap(err, "failed to delete existing AmneziaWG device") + } + case !netlinkAPI.IsLinkNotFoundError(err): + return errors.Wrapf(err, "error checking for existing AmneziaWG device %q", DefaultDeviceName) + } + + var userspaceDev UserspaceDevice + + userspaceDev, err = StartUserspaceDevice(DefaultDeviceName) + if err != nil { + return err + } + + a.userspaceDevice = userspaceDev + + link, err = a.netLink.LinkByName(DefaultDeviceName) + if err != nil { + return errors.Wrapf(err, "failed to find AmneziaWG link %s", DefaultDeviceName) + } + + a.link = link + + return nil +} + +func (a *amneziawgDriver) removePeer(key *wgtypes.Key) error { + logger.V(log.DEBUG).Infof("Removing AmneziaWG peer with key %s", key) + + peerCfg := []wgtypes.PeerConfig{ + { + PublicKey: *key, + Remove: true, + }, + } + + err := a.client.ConfigureDevice(DefaultDeviceName, wgtypes.Config{ + ReplacePeers: false, + Peers: peerCfg, + }) + + return errors.Wrapf(err, "failed to remove AmneziaWG peer with key %s", key) +} + +func (a *amneziawgDriver) peerByKey(key *wgtypes.Key) (*wgtypes.Peer, error) { + d, err := a.client.Device(DefaultDeviceName) + if err != nil { + return nil, errors.Wrapf(err, "failed to find device %s", DefaultDeviceName) + } + + for i := range d.Peers { + if d.Peers[i].PublicKey.String() == key.String() { + return &d.Peers[i], nil + } + } + + return nil, fmt.Errorf("peer not found for key %s", key) +} + +func (a *amneziawgDriver) verifyNewPeer(peerCfg *wgtypes.PeerConfig) error { + p, err := a.peerByKey(&peerCfg.PublicKey) + if err != nil { + return err + } + + if p.PresharedKey.String() != peerCfg.PresharedKey.String() { + return errors.New("peer PresharedKey does not match configured value") + } + + if p.Endpoint.String() != peerCfg.Endpoint.String() { + return fmt.Errorf("peer's Endpoint %q does not match configured %q", p.Endpoint.String(), peerCfg.Endpoint.String()) + } + + if !slices.EqualFunc(p.AllowedIPs, peerCfg.AllowedIPs, func(ipn1 net.IPNet, ipn2 net.IPNet) bool { + return ipn1.String() == ipn2.String() + }) { + return fmt.Errorf("peer's AllowedIPs %v does not match configured %q", p.AllowedIPs, peerCfg.AllowedIPs) + } + + logger.V(log.DEBUG).Infof("Peer configured, PublicKey: %s, EndPoint: %s, AllowedIPs: %v", p.PublicKey, p.Endpoint, p.AllowedIPs) + + return nil +} + +func (a *amneziawgDriver) keyMismatch(cid string, key *wgtypes.Key) bool { + c, found := a.connections[cid] + if !found { + logger.Warningf("Could not find spec for cluster %s, mismatched endpoint key %s", cid, key) + return true + } + + oldKey, _ := keyFromSpec(&c.Endpoint) + if oldKey.String() != key.String() { + logger.Warningf("Key mismatch, cluster %s key is %s, endpoint key is %s", cid, oldKey, key) + return true + } + + return false +} + +func genPsk(psk string) (wgtypes.Key, error) { + pskBytes := sha256.Sum256([]byte(psk)) + return wgtypes.NewKey(pskBytes[:]) //nolint:wrapcheck // Let the caller wrap it +} + +func (a *amneziawgDriver) Cleanup(_ context.Context) error { + logger.Info("Uninstalling the amneziawg cable driver") + + return a.cleanupDevice() +} + +func (a *amneziawgDriver) cleanupDevice() error { + if a.userspaceDevice != nil { + if err := a.userspaceDevice.Close(); err != nil { + logger.Warningf("Error closing userspace AmneziaWG device: %v", err) + } + + a.userspaceDevice = nil + } + + link, err := a.netLink.LinkByName(DefaultDeviceName) + if netlinkAPI.IsLinkNotFoundError(err) { + return nil + } + + if err != nil { + return errors.Wrapf(err, "error retrieving the AmneziaWG interface %q", DefaultDeviceName) + } + + err = a.netLink.LinkDel(link) + + return errors.Wrapf(err, "failed to delete existing AmneziaWG device %q", DefaultDeviceName) +} diff --git a/pkg/cable/amneziawg/getconnections.go b/pkg/cable/amneziawg/getconnections.go new file mode 100644 index 000000000..0e4bc235c --- /dev/null +++ b/pkg/cable/amneziawg/getconnections.go @@ -0,0 +1,182 @@ +/* +SPDX-License-Identifier: Apache-2.0 + +Copyright Contributors to the Submariner project. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + +http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package amneziawg + +import ( + "strconv" + "time" + + "github.com/advanced-wg/awgctrl-go/wgtypes" + "github.com/pkg/errors" + v1 "github.com/submariner-io/submariner/pkg/apis/submariner.io/v1" + "github.com/submariner-io/submariner/pkg/cable" + k8snet "k8s.io/utils/net" +) + +func (a *amneziawgDriver) GetConnections() ([]v1.Connection, error) { + d, err := a.client.Device(DefaultDeviceName) + if err != nil { + return nil, errors.Wrapf(err, "failed to find device %s", DefaultDeviceName) + } + + connections := make([]v1.Connection, 0) + + for i := range d.Peers { + key := d.Peers[i].PublicKey + + connection := a.connectionByKey(&key) + if connection == nil { + logger.Warningf("Found unknown peer with key %s, removing", key) + + if err := a.removePeer(&key); err != nil { + logger.Errorf(err, "Could not delete AmneziaWG peer with key %s, ignoring", key) + } + + continue + } + + a.updateConnectionForPeer(&d.Peers[i], connection) + connections = append(connections, *connection.DeepCopy()) + } + + return connections, nil +} + +func (a *amneziawgDriver) connectionByKey(key *wgtypes.Key) *v1.Connection { + for i := range a.connections { + // Since the endpoint was added to the connections list, it must have a valid public key, so we can + // safely ignore the error. + k, _ := keyFromSpec(&a.connections[i].Endpoint) + if key.String() == k.String() { + return a.connections[i] + } + } + + return nil +} + +// Update logic, based on delta from last check: good state requires a handshake and traffic; +// no handshake or a stale handshake moves the connection toward Connecting / ConnectionError. +func (a *amneziawgDriver) updateConnectionForPeer(p *wgtypes.Peer, connection *v1.Connection) { + now := int64(time.Nanosecond) * time.Now().UnixNano() / int64(time.Millisecond) + + pollMS := StatusPollInterval.Milliseconds() + 2 // +2 for rounding + + lc := peerTrafficDelta(connection, lastChecked, now) + if lc < pollMS { + // too fast to see any change, leave status as is + return + } + + // Expect Rx/Tx within the peer keepalive; if disabled, use the poll interval. + trafficWindow := a.keepAlive + if trafficWindow <= 0 { + trafficWindow = StatusPollInterval + } + + trafficMS := trafficWindow.Milliseconds() + 2 + + tx := peerTrafficDelta(connection, transmitBytes, p.TransmitBytes) + rx := peerTrafficDelta(connection, receiveBytes, p.ReceiveBytes) + lcSec := time.Duration(int64(time.Millisecond) * lc).Seconds() + + connectionFamily := connection.GetFamily() + + if p.LastHandshakeTime.IsZero() { + if lc > HandshakeTimeout.Milliseconds() { + // No initial handshake for too long. + connection.SetStatus(v1.ConnectionError, "no initial handshake for %.1f seconds", lcSec) + cable.RecordConnection(cableDriverName, &a.localEndpoint, &connection.Endpoint, string(connection.Status), false, connectionFamily) + } else if tx > 0 || rx > 0 { + // No handshake, but at least some communication in progress. + connection.SetStatus(v1.Connecting, "no initial handshake yet") + cable.RecordConnection(cableDriverName, &a.localEndpoint, &connection.Endpoint, string(connection.Status), false, connectionFamily) + } + + return + } + + if tx > 0 || rx > 0 { + // All is good. + connection.SetStatus(v1.Connected, "Rx=%d Bytes, Tx=%d Bytes", p.ReceiveBytes, p.TransmitBytes) + cable.RecordConnection(cableDriverName, &a.localEndpoint, &connection.Endpoint, string(connection.Status), false, connectionFamily) + saveAndRecordPeerTraffic(&a.localEndpoint, &connection.Endpoint, now, p.TransmitBytes, p.ReceiveBytes, connectionFamily) + + return + } + + handshakeDelta := time.Since(p.LastHandshakeTime) + + if handshakeDelta > HandshakeTimeout { + // Hard error, really long time since handshake. + connection.SetStatus(v1.ConnectionError, "no handshake for %.1f seconds", + handshakeDelta.Seconds()) + cable.RecordConnection(cableDriverName, &a.localEndpoint, &connection.Endpoint, string(connection.Status), false, connectionFamily) + + return + } + + if lc < 2*trafficMS { + // Grace period until keepalive (or poll floor) should have produced traffic. + logger.Warningf("No traffic for %.1f seconds; handshake was %.1f seconds ago: %#v", lcSec, handshakeDelta.Seconds(), connection) + return + } + + // Soft error, no traffic, stale handshake. + connection.SetStatus(v1.ConnectionError, "no bytes sent or received for %.1f seconds", + lcSec) + cable.RecordConnection(cableDriverName, &a.localEndpoint, &connection.Endpoint, string(connection.Status), false, connectionFamily) +} + +func (a *amneziawgDriver) updatePeerStatus(c *v1.Connection, key *wgtypes.Key) { + p, err := a.peerByKey(key) + if err != nil { + c.SetStatus(v1.ConnectionError, "cannot fetch status for peer %s: %v", key, err) + cable.RecordConnection(cableDriverName, &a.localEndpoint, &c.Endpoint, string(c.Status), false, c.GetFamily()) + + return + } + + a.updateConnectionForPeer(p, c) +} + +// Compare backendConfig[key] or initialize entry. +// Return delta from previous value or 0 if no previous value. +func peerTrafficDelta(c *v1.Connection, key string, newVal int64) int64 { + s, found := c.Endpoint.BackendConfig[key] + if found { + if i, err := strconv.ParseInt(s, 10, 64); err == nil { + return newVal - i + } + } + + c.Endpoint.BackendConfig[key] = strconv.FormatInt(newVal, 10) + + return 0 +} + +// Save backendConfig[key] and export the metrics to prometheus. +func saveAndRecordPeerTraffic(localEndpoint, remoteEndpoint *v1.EndpointSpec, lc, tx, rx int64, ipFamily k8snet.IPFamily) { + remoteEndpoint.BackendConfig[lastChecked] = strconv.FormatInt(lc, 10) + remoteEndpoint.BackendConfig[transmitBytes] = strconv.FormatInt(tx, 10) + remoteEndpoint.BackendConfig[receiveBytes] = strconv.FormatInt(rx, 10) + + cable.RecordTxBytes(cableDriverName, localEndpoint, remoteEndpoint, int(tx), ipFamily) + cable.RecordRxBytes(cableDriverName, localEndpoint, remoteEndpoint, int(rx), ipFamily) +} diff --git a/pkg/cable/amneziawg/obfuscation.go b/pkg/cable/amneziawg/obfuscation.go new file mode 100644 index 000000000..8596912c5 --- /dev/null +++ b/pkg/cable/amneziawg/obfuscation.go @@ -0,0 +1,245 @@ +/* +SPDX-License-Identifier: Apache-2.0 + +Copyright Contributors to the Submariner project. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + +http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package amneziawg + +import ( + "encoding/hex" + "strconv" + "strings" + + "github.com/advanced-wg/awgctrl-go/wgtypes" + "github.com/pkg/errors" +) + +// obfuscationOptionTable is the AmneziaWG obfuscation options and built-in defaults. +// +// AmneziaWG always applies these defaults (there is no "plain WireGuard" mode via options): +// the driver is intentionally an obfuscating tunnel, not a WG drop-in with optional junk. +// +// H1–H4 and S1–S4 MUST be identical on every Submariner peer (every cluster using this driver). +// Jc / Jmin / Jmax / I1–I5 may differ per side, but the shipped defaults keep both sides symmetric +// so a single cableDriverOptions map (or no overrides) works out of the box. Mismatched H*/S* +// typically yields silent connectivity failure — override carefully and keep clusters in sync. +func obfuscationOptionTable(cfg *wgtypes.Config) []cableDriverOption { + return []cableDriverOption{ + // Junk packets before each handshake (helps break WG size signatures). + option("jc", "7", parsePositiveInt, &cfg.Jc), + option("jmin", "80", parsePositiveInt, &cfg.Jmin), + option("jmax", "200", parsePositiveInt, &cfg.Jmax), + + // Message padding (AWG 2.0: S3 cookie, S4 transport). Must match on every peer. + option("s1", "45", parseNonNegativeInt, &cfg.S1), + option("s2", "60", parseNonNegativeInt, &cfg.S2), + option("s3", "35", parseNonNegativeInt, &cfg.S3), + option("s4", "12", parseNonNegativeInt, &cfg.S4), + + // Magic headers as non-overlapping ranges (AWG 2.0) — not the fixed WG types 1–4. + // Must match on every peer. + option("h1", "200000000-280000000", parseHeader, &cfg.H1), + option("h2", "400000000-480000000", parseHeader, &cfg.H2), + option("h3", "600000000-680000000", parseHeader, &cfg.H3), + option("h4", "800000000-880000000", parseHeader, &cfg.H4), + + // Custom init packets (AWG 2.0) — random-looking UDP before handshake. + option("i1", "", parseInitPacket, &cfg.I1), + option("i2", "", parseInitPacket, &cfg.I2), + option("i3", "", parseInitPacket, &cfg.I3), + option("i4", "", parseInitPacket, &cfg.I4), + option("i5", "", parseInitPacket, &cfg.I5), + } +} + +func validateObfuscationOptions(table []cableDriverOption) []error { + var errs []error + + jmin, jminOK := optionValue[*int](table, "jmin") + jmax, jmaxOK := optionValue[*int](table, "jmax") + + if jminOK && jmaxOK && jmin != nil && jmax != nil && *jmin > *jmax { + errs = append(errs, errors.Errorf( + "invalid AmneziaWG options: jmin (%d) must be <= jmax (%d)", *jmin, *jmax)) + } + + type namedRange struct { + name string + value string + r headerRange + } + + headers := make([]namedRange, 0, 4) + + for _, key := range []string{"h1", "h2", "h3", "h4"} { + val, ok := optionValue[*string](table, key) + if !ok || val == nil { + continue + } + + r, err := parseHeaderRange(*val) + if err != nil { + errs = append(errs, errors.Wrapf(err, "invalid AmneziaWG option %q", key)) + continue + } + + headers = append(headers, namedRange{name: key, value: *val, r: r}) + } + + for i := range headers { + for j := i + 1; j < len(headers); j++ { + left, right := headers[i], headers[j] + if left.r.start <= right.r.end && right.r.start <= left.r.end { + errs = append(errs, errors.Errorf( + "invalid AmneziaWG options: %s (%s) overlaps %s (%s)", + left.name, left.value, right.name, right.value)) + } + } + } + + return errs +} + +type headerRange struct { + start uint64 + end uint64 +} + +func parseHeader(key, value string) (*string, error) { + if _, err := parseHeaderRange(value); err != nil { + return nil, errors.Wrapf(err, "invalid AmneziaWG option %q=%q", key, value) + } + + return new(value), nil +} + +// parseHeaderRange mirrors amneziawg-go newMagicHeader: single uint32 or min-max with min <= max. +func parseHeaderRange(value string) (headerRange, error) { + parts := strings.Split(value, "-") + if len(parts) < 1 || len(parts) > 2 { + return headerRange{}, errors.New("expected uint32 or min-max range") + } + + start, err := strconv.ParseUint(parts[0], 10, 32) + if err != nil { + return headerRange{}, errors.Wrap(err, "invalid range start") + } + + end := start + if len(parts) == 2 { + end, err = strconv.ParseUint(parts[1], 10, 32) + if err != nil { + return headerRange{}, errors.Wrap(err, "invalid range end") + } + } + + if end < start { + return headerRange{}, errors.New("range start must be <= end") + } + + return headerRange{start: start, end: end}, nil +} + +func parseInitPacket(key, value string) (*string, error) { + if err := validateInitPacketSpec(value); err != nil { + return nil, errors.Wrapf(err, "invalid AmneziaWG option %q=%q", key, value) + } + + return new(value), nil +} + +// validateInitPacketSpec mirrors amneziawg-go newObfChain tag grammar closely enough to fail fast. +func validateInitPacketSpec(spec string) error { + remaining := spec + tagCount := 0 + + for { + start := strings.IndexByte(remaining, '<') + if start == -1 { + break + } + + endRel := strings.IndexByte(remaining[start:], '>') + if endRel == -1 { + return errors.New("missing enclosing '>'") + } + + end := start + endRel + tag := remaining[start+1 : end] + parts := strings.Fields(tag) + + if len(parts) == 0 { + return errors.New("empty tag") + } + + param := "" + if len(parts) > 1 { + param = parts[1] + } + + if err := validateInitPacketTag(parts[0], param); err != nil { + return err + } + + tagCount++ + remaining = remaining[end+1:] + } + + if tagCount == 0 { + return errors.New("expected one or more values") + } + + return nil +} + +func validateInitPacketTag(name, param string) error { + switch name { + case "b": + hexStr := strings.TrimPrefix(strings.TrimPrefix(param, "0x"), "0X") + if hexStr == "" { + return errors.Errorf("tag requires hex bytes") + } + + if len(hexStr)%2 != 0 { + return errors.Errorf("tag hex must have even length") + } + + if _, err := hex.DecodeString(hexStr); err != nil { + return errors.Wrap(err, "tag invalid hex") + } + + return nil + + case "r", "rc", "rd": + n, err := strconv.Atoi(param) + if err != nil { + return errors.Wrapf(err, "tag <%s> requires an integer size", name) + } + + if n < 0 { + return errors.Errorf("tag <%s> size must be >= 0", name) + } + + return nil + + case "t", "d", "ds", "dz": + // amneziawg-go accepts these with an ignored/empty parameter. + return nil + + default: + return errors.Errorf("unknown tag <%s>", name) + } +} diff --git a/pkg/cable/amneziawg/options.go b/pkg/cable/amneziawg/options.go new file mode 100644 index 000000000..f636e0a70 --- /dev/null +++ b/pkg/cable/amneziawg/options.go @@ -0,0 +1,189 @@ +/* +SPDX-License-Identifier: Apache-2.0 + +Copyright Contributors to the Submariner project. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + +http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package amneziawg + +import ( + stderrors "errors" + "sort" + "strconv" + "time" + + "github.com/advanced-wg/awgctrl-go/wgtypes" + "github.com/pkg/errors" + "github.com/submariner-io/submariner/pkg/cable" +) + +// cableDriverOption is one SUBMARINER_CABLEDRIVEROPTIONS key. +type cableDriverOption struct { + key string // JSON option name (e.g. "jc", "h1", "keepalive") + def string // default when the key is absent or empty in the env map + dest any // *T — filled by parse, then read back for cross-field validation + parse func(key, value string) (any, error) // parses value, stores into dest, returns the parsed value +} + +// cableDriverOptionTable is the full set of keys accepted in cableDriverOptions. +func cableDriverOptionTable(a *amneziawgDriver, cfg *wgtypes.Config) []cableDriverOption { + return append(obfuscationOptionTable(cfg), driverOptionTable(a)...) +} + +// driverOptionTable holds non-obfuscation driver knobs from cableDriverOptions. +func driverOptionTable(a *amneziawgDriver) []cableDriverOption { + return []cableDriverOption{ + // Persistent keepalive for peers (seconds). 0 disables; useful behind strict NAT. + option("keepalive", "10", parseKeepAliveSeconds, &a.keepAlive), + } +} + +// option binds a JSON key to dest. parse validates and converts the string value. +func option[T any](key, def string, parse func(key, value string) (T, error), dest *T) cableDriverOption { + return cableDriverOption{ + key: key, + def: def, + dest: dest, + parse: func(key, value string) (any, error) { + v, err := parse(key, value) + if err != nil { + return nil, err + } + + *dest = v + + return v, nil + }, + } +} + +func (o *cableDriverOption) set(value string) error { + _, err := o.parse(o.key, value) + + return err +} + +func applyCableDriverOptions(a *amneziawgDriver, cfg *wgtypes.Config) error { + options, err := cable.GetDriverOptions() + if err != nil { + return errors.Wrap(err, "error reading cable driver options") + } + + return applyCableDriverOptionsMap(a, cfg, options) +} + +// applyCableDriverOptionsMap applies obfuscation + driver option tables. +// Configuration problems are collected and returned together for a single fix pass. +func applyCableDriverOptionsMap(a *amneziawgDriver, cfg *wgtypes.Config, options map[string]string) error { + table := cableDriverOptionTable(a, cfg) + + known := make(map[string]struct{}, len(table)) + for _, opt := range table { + known[opt.key] = struct{}{} + } + + var errs []error + + unknown := make([]string, 0) + + for key := range options { + if _, ok := known[key]; !ok { + unknown = append(unknown, key) + } + } + + sort.Strings(unknown) + + for _, key := range unknown { + errs = append(errs, errors.Errorf("unknown AmneziaWG option %q", key)) + } + + for i := range table { + value := table[i].def + if v, ok := options[table[i].key]; ok && v != "" { + value = v + } + + if err := table[i].set(value); err != nil { + errs = append(errs, err) + } + } + + errs = append(errs, validateObfuscationOptions(table)...) + + return stderrors.Join(errs...) +} + +func optionValue[T any](table []cableDriverOption, key string) (T, bool) { + var zero T + + for i := range table { + if table[i].key != key { + continue + } + + dest, ok := table[i].dest.(*T) + if !ok || dest == nil { + return zero, false + } + + return *dest, true + } + + return zero, false +} + +func parsePositiveInt(key, value string) (*int, error) { + n, err := strconv.Atoi(value) + if err != nil { + return nil, errors.Wrapf(err, "invalid AmneziaWG option %q=%q", key, value) + } + + if n <= 0 { + return nil, errors.Errorf("invalid AmneziaWG option %q=%q: must be > 0", key, value) + } + + return new(n), nil +} + +func parseNonNegativeInt(key, value string) (*int, error) { + n, err := strconv.Atoi(value) + if err != nil { + return nil, errors.Wrapf(err, "invalid AmneziaWG option %q=%q", key, value) + } + + if n < 0 { + return nil, errors.Errorf("invalid AmneziaWG option %q=%q: must be >= 0", key, value) + } + + return new(n), nil +} + +// WireGuard persistent-keepalive is stored as seconds in a uint16-sized field. +const maxKeepAliveSeconds = 65535 + +func parseKeepAliveSeconds(key, value string) (time.Duration, error) { + sec, err := strconv.Atoi(value) + if err != nil { + return 0, errors.Wrapf(err, "invalid AmneziaWG option %q=%q", key, value) + } + + if sec < 0 || sec > maxKeepAliveSeconds { + return 0, errors.Errorf( + "invalid AmneziaWG option %q=%q: must be between 0 and %d seconds", key, value, maxKeepAliveSeconds) + } + + return time.Duration(sec) * time.Second, nil +} diff --git a/pkg/cable/amneziawg/options_internal_test.go b/pkg/cable/amneziawg/options_internal_test.go new file mode 100644 index 000000000..103c8e4ed --- /dev/null +++ b/pkg/cable/amneziawg/options_internal_test.go @@ -0,0 +1,211 @@ +/* +SPDX-License-Identifier: Apache-2.0 + +Copyright Contributors to the Submariner project. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + +http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package amneziawg + +import ( + "time" + + "github.com/advanced-wg/awgctrl-go/wgtypes" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +func newOptionsDriver() *amneziawgDriver { + return &amneziawgDriver{keepAlive: 10 * time.Second} +} + +var _ = Describe("parseHeaderRange", func() { + DescribeTable("validation", + func(value string, wantErr bool) { + _, err := parseHeaderRange(value) + if wantErr { + Expect(err).To(HaveOccurred()) + } else { + Expect(err).NotTo(HaveOccurred()) + } + }, + Entry("single uint32", "123", false), + Entry("valid range", "10-20", false), + Entry("equal bounds", "5-5", false), + Entry("inverted range", "20-10", true), + Entry("non-numeric", "foo", true), + Entry("too many parts", "1-2-3", true), + Entry("empty", "", true), + Entry("trailing dash", "10-", true), + Entry("leading dash", "-10", true), + ) +}) + +var _ = Describe("validateInitPacketSpec", func() { + DescribeTable("validation", + func(value string, wantErr bool) { + err := validateInitPacketSpec(value) + if wantErr { + Expect(err).To(HaveOccurred()) + } else { + Expect(err).NotTo(HaveOccurred()) + } + }, + Entry("random tag", "", false), + Entry("hex tag", "", false), + Entry("hex without 0x", "", false), + Entry("combined tags", "", false), + Entry("timestamp tag", "", false), + Entry("unknown tag", "", true), + Entry("plain hex", "abcd", true), + Entry("empty", "", true), + Entry("empty tags", "<>", true), + Entry("unclosed tag", "", true), + Entry("odd hex", "", true), + ) +}) + +var _ = Describe("applyCableDriverOptionsMap", func() { + It("should apply defaults when options are empty", func() { + a := newOptionsDriver() + var cfg wgtypes.Config + Expect(applyCableDriverOptionsMap(a, &cfg, nil)).To(Succeed()) + + Expect(*cfg.Jc).To(Equal(7)) + Expect(*cfg.H1).To(Equal("200000000-280000000")) + Expect(*cfg.I1).To(Equal("")) + Expect(*cfg.S3).To(Equal(35)) + Expect(a.keepAlive).To(Equal(10 * time.Second)) + }) + + It("should apply valid overrides and keep untouched defaults", func() { + a := newOptionsDriver() + var cfg wgtypes.Config + Expect(applyCableDriverOptionsMap(a, &cfg, map[string]string{ + "jc": "3", + "h1": "10-20", + "i1": "", + "keepalive": "25", + })).To(Succeed()) + + Expect(*cfg.Jc).To(Equal(3)) + Expect(*cfg.H1).To(Equal("10-20")) + Expect(*cfg.I1).To(Equal("")) + Expect(*cfg.S3).To(Equal(35)) + Expect(*cfg.H2).To(Equal("400000000-480000000")) + Expect(a.keepAlive).To(Equal(25 * time.Second)) + }) + + It("should fail on an unknown option key", func() { + err := applyCableDriverOptionsMap(newOptionsDriver(), &wgtypes.Config{}, map[string]string{"xyz": "1"}) + Expect(err).To(MatchError(ContainSubstring(`unknown AmneziaWG option "xyz"`))) + }) + + It("should fail on a mistyped option key", func() { + err := applyCableDriverOptionsMap(newOptionsDriver(), &wgtypes.Config{}, map[string]string{"Jc": "3"}) + Expect(err).To(MatchError(ContainSubstring(`unknown AmneziaWG option "Jc"`))) + }) + + It("should report multiple option errors together", func() { + err := applyCableDriverOptionsMap(newOptionsDriver(), &wgtypes.Config{}, map[string]string{ + "xyz": "1", + "jc": "0", + "h1": "20-10", + }) + Expect(err).To(HaveOccurred()) + Expect(err.Error()).To(And( + ContainSubstring(`unknown AmneziaWG option "xyz"`), + ContainSubstring(`invalid AmneziaWG option "jc"="0"`), + ContainSubstring(`invalid AmneziaWG option "h1"="20-10"`), + )) + }) + + It("should keep defaults for empty option values", func() { + a := newOptionsDriver() + var cfg wgtypes.Config + Expect(applyCableDriverOptionsMap(a, &cfg, map[string]string{ + "jc": "", + "h1": "", + "i1": "", + "keepalive": "", + })).To(Succeed()) + + Expect(*cfg.Jc).To(Equal(7)) + Expect(*cfg.H1).To(Equal("200000000-280000000")) + Expect(*cfg.I1).To(Equal("")) + Expect(a.keepAlive).To(Equal(10 * time.Second)) + }) + + It("should fail on a non-positive junk count", func() { + err := applyCableDriverOptionsMap(newOptionsDriver(), &wgtypes.Config{}, map[string]string{"jc": "0"}) + Expect(err).To(MatchError(ContainSubstring(`invalid AmneziaWG option "jc"="0"`))) + }) + + It("should fail on an invalid integer option", func() { + err := applyCableDriverOptionsMap(newOptionsDriver(), &wgtypes.Config{}, map[string]string{"jc": "not-a-number"}) + Expect(err).To(MatchError(ContainSubstring(`invalid AmneziaWG option "jc"="not-a-number"`))) + }) + + It("should fail on an invalid header option", func() { + err := applyCableDriverOptionsMap(newOptionsDriver(), &wgtypes.Config{}, map[string]string{"h1": "20-10"}) + Expect(err).To(MatchError(ContainSubstring(`invalid AmneziaWG option "h1"="20-10"`))) + }) + + It("should fail when header ranges overlap", func() { + err := applyCableDriverOptionsMap(newOptionsDriver(), &wgtypes.Config{}, map[string]string{ + "h1": "10-30", + "h2": "20-40", + }) + Expect(err).To(MatchError(ContainSubstring("overlaps"))) + }) + + It("should fail on an invalid init packet option", func() { + err := applyCableDriverOptionsMap(newOptionsDriver(), &wgtypes.Config{}, map[string]string{"i1": ""}) + Expect(err).To(MatchError(ContainSubstring(`invalid AmneziaWG option "i1"=""`))) + }) + + It("should fail on an invalid keepalive option", func() { + err := applyCableDriverOptionsMap(newOptionsDriver(), &wgtypes.Config{}, map[string]string{"keepalive": "-1"}) + Expect(err).To(MatchError(ContainSubstring(`invalid AmneziaWG option "keepalive"="-1"`))) + }) + + It("should fail when keepalive exceeds the WireGuard maximum", func() { + err := applyCableDriverOptionsMap(newOptionsDriver(), &wgtypes.Config{}, map[string]string{"keepalive": "65536"}) + Expect(err).To(MatchError(ContainSubstring(`invalid AmneziaWG option "keepalive"="65536"`))) + }) + + It("should allow disabling keepalive with 0", func() { + a := newOptionsDriver() + Expect(applyCableDriverOptionsMap(a, &wgtypes.Config{}, map[string]string{"keepalive": "0"})).To(Succeed()) + Expect(a.keepAlive).To(Equal(time.Duration(0))) + }) + + It("should fail when jmin is greater than jmax", func() { + err := applyCableDriverOptionsMap(newOptionsDriver(), &wgtypes.Config{}, map[string]string{ + "jmin": "200", + "jmax": "80", + }) + Expect(err).To(MatchError(ContainSubstring("jmin (200) must be <= jmax (80)"))) + }) + + It("should not clobber non-obfuscation config fields", func() { + port := 51820 + cfg := wgtypes.Config{ListenPort: &port} + + Expect(applyCableDriverOptionsMap(newOptionsDriver(), &cfg, map[string]string{"jc": "3"})).To(Succeed()) + Expect(cfg.ListenPort).To(Equal(&port)) + Expect(*cfg.Jc).To(Equal(3)) + }) +}) diff --git a/pkg/cable/cable_suite_test.go b/pkg/cable/cable_suite_test.go new file mode 100644 index 000000000..7c28cf9bb --- /dev/null +++ b/pkg/cable/cable_suite_test.go @@ -0,0 +1,31 @@ +/* +SPDX-License-Identifier: Apache-2.0 + +Copyright Contributors to the Submariner project. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + +http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package cable_test + +import ( + "testing" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +func TestCable(t *testing.T) { + RegisterFailHandler(Fail) + RunSpecs(t, "Cable Suite") +} diff --git a/pkg/cable/options.go b/pkg/cable/options.go new file mode 100644 index 000000000..850e46931 --- /dev/null +++ b/pkg/cable/options.go @@ -0,0 +1,46 @@ +/* +SPDX-License-Identifier: Apache-2.0 + +Copyright Contributors to the Submariner project. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + +http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package cable + +import ( + "encoding/json" + "os" + + "github.com/pkg/errors" +) + +// CableDriverOptionsEnv is set by the operator from Submariner.spec.cableDriverOptions. +const CableDriverOptionsEnv = "SUBMARINER_CABLEDRIVEROPTIONS" + +// GetDriverOptions returns driver-specific options from the environment. +// An empty value yields an empty map. Invalid JSON is returned as an error so +// misconfiguration fails fast instead of being silently ignored. +func GetDriverOptions() (map[string]string, error) { + raw := os.Getenv(CableDriverOptionsEnv) + if raw == "" { + return map[string]string{}, nil + } + + options := map[string]string{} + if err := json.Unmarshal([]byte(raw), &options); err != nil { + return nil, errors.Wrapf(err, "error parsing %q", CableDriverOptionsEnv) + } + + return options, nil +} diff --git a/pkg/cable/options_test.go b/pkg/cable/options_test.go new file mode 100644 index 000000000..043bcddb2 --- /dev/null +++ b/pkg/cable/options_test.go @@ -0,0 +1,66 @@ +/* +SPDX-License-Identifier: Apache-2.0 + +Copyright Contributors to the Submariner project. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + +http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package cable_test + +import ( + "os" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + "github.com/submariner-io/submariner/pkg/cable" +) + +var _ = Describe("GetDriverOptions", func() { + AfterEach(func() { + Expect(os.Unsetenv(cable.CableDriverOptionsEnv)).To(Succeed()) + }) + + When("the environment variable is unset or empty", func() { + It("should return an empty map", func() { + Expect(os.Setenv(cable.CableDriverOptionsEnv, "")).To(Succeed()) + + options, err := cable.GetDriverOptions() + Expect(err).NotTo(HaveOccurred()) + Expect(options).To(BeEmpty()) + }) + }) + + When("the environment variable contains valid JSON", func() { + It("should return the parsed options", func() { + Expect(os.Setenv(cable.CableDriverOptionsEnv, `{"jc":"9","h1":"1-2"}`)).To(Succeed()) + + options, err := cable.GetDriverOptions() + Expect(err).NotTo(HaveOccurred()) + Expect(options).To(Equal(map[string]string{ + "jc": "9", + "h1": "1-2", + })) + }) + }) + + When("the environment variable contains invalid JSON", func() { + It("should return an error", func() { + Expect(os.Setenv(cable.CableDriverOptionsEnv, `{not-json`)).To(Succeed()) + + options, err := cable.GetDriverOptions() + Expect(err).To(HaveOccurred()) + Expect(options).To(BeNil()) + }) + }) +}) diff --git a/pkg/cableengine/cableengine.go b/pkg/cableengine/cableengine.go index 9b6f13b39..f885673c0 100644 --- a/pkg/cableengine/cableengine.go +++ b/pkg/cableengine/cableengine.go @@ -37,6 +37,7 @@ import ( logf "sigs.k8s.io/controller-runtime/pkg/log" // Add supported drivers. + _ "github.com/submariner-io/submariner/pkg/cable/amneziawg" _ "github.com/submariner-io/submariner/pkg/cable/libreswan" _ "github.com/submariner-io/submariner/pkg/cable/vxlan" _ "github.com/submariner-io/submariner/pkg/cable/wireguard"