Integrate startup-verified IPAM into the node agent

api/machina/v1alpha3/site_types.go

@@ -67,10 +67,10 @@ type SiteSpec struct {
 	// CNI config is written and all same-site nodes are tunnel peers.
 	// When false, only tunnel links to gateway pools for other sites are created,
 	// allowing an external CNI plugin to manage intra-site networking.
-	// Pod CIDR assignment is also disabled when manageCniPlugin is false,
-	// regardless of the assignmentEnabled setting on individual podCidrAssignments
-	// rules. The podCidrAssignments are still required to define the CIDR ranges
-	// used for inter-site routing.
+	// Omitted assignmentEnabled values on podCidrAssignments default to this
+	// setting. Explicit assignmentEnabled values override that default.
+	// The podCidrAssignments are still required to define the CIDR ranges
+	// used for inter-site routing, even when assignment is disabled.
 	// +optional
 	ManageCniPlugin *bool `json:"manageCniPlugin,omitempty"`
 
@@ -107,6 +107,16 @@ type SiteSpec struct {
 	Components SiteComponents `json:"components,omitempty"`
 }
 
+// PodCIDRAssignmentEnabled resolves an assignment's explicit setting or the
+// site's CNI management default without persisting an inherited value.
+func (s SiteSpec) PodCIDRAssignmentEnabled(assignment unboundednetv1alpha1.PodCidrAssignment) bool {
+	if assignment.AssignmentEnabled != nil {
+		return *assignment.AssignmentEnabled
+	}
+
+	return s.ManageCniPlugin == nil || *s.ManageCniPlugin
+}
+
 // SiteComponents declares optional components managed by the unbounded operator.
 // Networking (unbounded-net) is not a component: it is a cluster singleton the
 // operator deploys whenever at least one Site exists.

api/machina/v1alpha3/site_types_test.go

@@ -36,6 +36,47 @@ func TestSiteResourceAndAddToScheme(t *testing.T) {
 	}
 }
 
+func TestPodCIDRAssignmentEnabled(t *testing.T) {
+	settings := []struct {
+		name  string
+		value *bool
+	}{
+		{"omitted", nil},
+		{"true", new(true)},
+		{"false", new(false)},
+	}
+	for _, manage := range settings {
+		for _, assignment := range settings {
+			t.Run(manage.name+"/"+assignment.name, func(t *testing.T) {
+				spec := SiteSpec{ManageCniPlugin: manage.value}
+				rule := unboundednetv1alpha1.PodCidrAssignment{AssignmentEnabled: assignment.value}
+
+				want := assignment.name == "true" || (assignment.name == "omitted" && manage.name != "false")
+				if got := spec.PodCIDRAssignmentEnabled(rule); got != want {
+					t.Fatalf("effective assignment enabled = %v, want %v", got, want)
+				}
+
+				if rule.AssignmentEnabled != assignment.value {
+					t.Fatal("resolution mutated the assignment setting")
+				}
+			})
+		}
+	}
+
+	spec := SiteSpec{}
+	rule := unboundednetv1alpha1.PodCidrAssignment{}
+
+	spec.ManageCniPlugin = new(false)
+	if spec.PodCIDRAssignmentEnabled(rule) {
+		t.Fatal("omitted assignment did not inherit changed CNI setting")
+	}
+
+	spec.ManageCniPlugin = new(true)
+	if !spec.PodCIDRAssignmentEnabled(rule) || rule.AssignmentEnabled != nil {
+		t.Fatal("inherited default was persisted instead of resolved")
+	}
+}
+
 func TestDeepCopySiteAndList(t *testing.T) {
 	enabled := true
 	priority := int32(10)

api/net/v1alpha1/types.go

@@ -36,7 +36,9 @@ const (
 // PodCidrAssignment defines a pod CIDR allocation rule for nodes in a site.
 // +k8s:deepcopy-gen=true
 type PodCidrAssignment struct {
-	// AssignmentEnabled controls whether this assignment is active. Defaults to true.
+	// AssignmentEnabled controls whether this assignment is active. When omitted,
+	// it inherits the Site's manageCniPlugin setting, which defaults to true.
+	// Explicit true or false overrides the inherited value.
 	// +optional
 	AssignmentEnabled *bool `json:"assignmentEnabled,omitempty"`
 

cmd/unbounded-net-controller/node_auth_informers_test.go

@@ -18,6 +18,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/watch"
 	k8sfake "k8s.io/client-go/kubernetes/fake"
 	k8stesting "k8s.io/client-go/testing"
 	"k8s.io/client-go/tools/cache"
@@ -129,8 +130,28 @@ func TestNodeAuthInformersScopeAndLifecycle(t *testing.T) {
 	otherSA := sa.DeepCopy()
 	otherSA.Namespace = "other"
 	client := k8sfake.NewClientset(pod, sa, unlabeled, otherPod, otherSA)
+	podWatchStarted := make(chan struct{}, 1)
+	watchReactor := client.WatchReactionChain[0]
+	client.PrependWatchReactor("pods", func(action k8stesting.Action) (bool, watch.Interface, error) {
+		handled, stream, err := watchReactor.React(action)
+		if handled && err == nil {
+			select {
+			case podWatchStarted <- struct{}{}:
+			default:
+			}
+		}
+
+		return handled, stream, err
+	})
 	caches, stop := startNodeAuthTestCaches(t, client)
 
+	// The fake tracker cannot replay deletions between the initial list and watch registration.
+	select {
+	case <-podWatchStarted:
+	case <-time.After(5 * time.Second):
+		t.Fatal("Pod watch did not start")
+	}
+
 	if _, err := caches.pods.Lister().Pods(pod.Namespace).Get(pod.Name); err != nil {
 		t.Fatal(err)
 	}

cmd/unbounded-net-node/bootstrap_helpers.go

@@ -24,6 +24,7 @@ import (
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/klog/v2"
 
+	"github.com/Azure/unbounded/internal/net/ipam"
 	"github.com/Azure/unbounded/internal/version"
 )
 
@@ -78,6 +79,10 @@ func waitForPodCIDRsAndConfigure(ctx context.Context, clientset kubernetes.Inter
 
 		// Check if podCIDRs are already assigned
 		if len(node.Spec.PodCIDRs) > 0 {
+			if _, err := ipam.ConfigFromCIDRs(node.Spec.PodCIDRs); err != nil {
+				return nil, err
+			}
+
 			klog.V(4).Infof("Node %s has podCIDRs: %v", cfg.NodeName, node.Spec.PodCIDRs)
 
 			if err := guardedWriteCNIConfig(ctx, cfg, node.Spec.PodCIDRs, healthState); err != nil {
@@ -137,6 +142,11 @@ func waitForPodCIDRsAndConfigure(ctx context.Context, clientset kubernetes.Inter
 					}
 
 					if len(updatedNode.Spec.PodCIDRs) > 0 {
+						if _, err := ipam.ConfigFromCIDRs(updatedNode.Spec.PodCIDRs); err != nil {
+							watcher.Stop()
+							return nil, err
+						}
+
 						klog.Infof("Node %s podCIDRs assigned: %v", cfg.NodeName, updatedNode.Spec.PodCIDRs)
 						watcher.Stop()
 
@@ -204,10 +214,9 @@ func writeCNIConfigUnchecked(cfg *config, podCIDRs []string) error {
 }
 
 func doWriteCNIConfig(cfg *config, podCIDRs []string) error {
-	// Build IPAM ranges from podCIDRs
-	var ranges [][]IPRange
-	for _, cidr := range podCIDRs {
-		ranges = append(ranges, []IPRange{{Subnet: cidr}})
+	socketPath := cfg.IPAMSocketPath
+	if socketPath == ipam.DefaultSocketPath {
+		socketPath = ""
 	}
 
 	// Bridge MTU matches the configured WireGuard tunnel MTU (cfg.MTU is
@@ -229,8 +238,8 @@ func doWriteCNIConfig(cfg *config, podCIDRs []string) error {
 				HairpinMode:  true,
 				MTU:          bridgeMTU,
 				IPAM: &IPAMConfig{
-					Type:   "host-local",
-					Ranges: ranges,
+					Type:       "unbounded-ipam",
+					SocketPath: socketPath,
 				},
 			},
 			{

cmd/unbounded-net-node/bootstrap_helpers_test.go

@@ -201,12 +201,12 @@ func TestWriteCNIConfigWritesExpectedConflist(t *testing.T) {
 		t.Fatalf("unexpected bridge plugin config: %#v", got.Plugins[0])
 	}
 
-	if len(got.Plugins[0].IPAM.Ranges) != 2 {
-		t.Fatalf("expected two IPAM ranges, got %d", len(got.Plugins[0].IPAM.Ranges))
+	if got.Plugins[0].IPAM.Type != "unbounded-ipam" || got.Plugins[0].IPAM.SocketPath != "" {
+		t.Fatalf("expected argument-free unbounded-ipam, got %#v", got.Plugins[0].IPAM)
 	}
 
-	if got.Plugins[0].IPAM.Ranges[0][0].Subnet != podCIDRs[0] || got.Plugins[0].IPAM.Ranges[1][0].Subnet != podCIDRs[1] {
-		t.Fatalf("unexpected IPAM ranges: %#v", got.Plugins[0].IPAM.Ranges)
+	if strings.Contains(string(raw), "10.244.0.0") || strings.Contains(string(raw), "fd00:10::") {
+		t.Fatalf("conflist contains cached PodCIDRs: %s", raw)
 	}
 
 	if _, err := os.Stat(confPath + ".tmp"); !os.IsNotExist(err) {

cmd/unbounded-net-node/cni_reconcile_test.go

@@ -7,6 +7,7 @@ import (
 	"context"
 	"errors"
 	"os"
+	"slices"
 	"testing"
 
 	corev1 "k8s.io/api/core/v1"
@@ -18,6 +19,80 @@ import (
 	unboundednetnetlink "github.com/Azure/unbounded/internal/net/netlink"
 )
 
+func TestDelegatedAssignmentReconcilesAfterEmptyBootstrap(t *testing.T) {
+	cfg := newCNISafetyTestConfig(t)
+	health := &nodeHealthState{}
+	health.markCNIUnmanaged()
+
+	state := &wireGuardState{
+		clientset:   fake.NewClientset(&corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: cfg.NodeName}}),
+		nodeName:    cfg.NodeName,
+		healthState: health,
+	}
+
+	originalConfigure := configureWireGuardFunc
+
+	t.Cleanup(func() { configureWireGuardFunc = originalConfigure })
+
+	var configured [][]string
+
+	fail := false
+	configureWireGuardFunc = func(_ context.Context, _ *config, _ string, _ []meshPeerInfo, _ []gatewayPeerInfo, _ string, _, _, _ map[string]bool, _, _, _, _, _ map[string]string, _, _, _, _, _ map[string]int, _ []unboundednetnetlink.DesiredRoute, _ map[string]bool, current *wireGuardState, _ map[string]healthcheck.HealthCheckSettings) error {
+		configured = append(configured, slices.Clone(current.nodePodCIDRs))
+
+		if fail {
+			return errors.New("injected routing failure")
+		}
+
+		return nil
+	}
+
+	sites := newInformerWithObjects(toUnstructured(t, &unboundedv1alpha3.Site{
+		ObjectMeta: metav1.ObjectMeta{Name: "site-a"},
+	}))
+	empty := newInformerWithObjects()
+	reconcile := func() error {
+		return updateWireGuardFromSlices(t.Context(), nil, sites, empty, empty, empty, empty, empty, empty, cfg, "site-a", "private", "public", false, state)
+	}
+
+	if err := reconcile(); err != nil {
+		t.Fatal(err)
+	}
+
+	cidrs := []string{"10.244.3.0/24", "fd00:3::/64"}
+	health.setVerifiedIPAMAssignment(cidrs)
+
+	fail = true
+
+	if err := reconcile(); err == nil {
+		t.Fatal("routing failure was not reported")
+	}
+
+	if len(state.nodePodCIDRs) != 0 {
+		t.Fatalf("failed routing committed assignment: %v", state.nodePodCIDRs)
+	}
+
+	fail = false
+
+	if err := reconcile(); err != nil {
+		t.Fatal(err)
+	}
+
+	if !slices.Equal(state.nodePodCIDRs, cidrs) || len(configured) == 0 || !slices.Equal(configured[len(configured)-1], cidrs) {
+		t.Fatalf("delayed assignment not configured: state=%v configured=%v", state.nodePodCIDRs, configured)
+	}
+
+	count := len(configured)
+
+	if err := reconcile(); err != nil {
+		t.Fatal(err)
+	}
+
+	if len(configured) != count {
+		t.Fatal("unchanged assignment unnecessarily reconfigured routing")
+	}
+}
+
 func TestCNIReconciliationDisablesAndRecoversWithoutMTUChange(t *testing.T) {
 	cfg := newCNISafetyTestConfig(t)
 	cidrs := []string{"10.244.2.0/24"}

cmd/unbounded-net-node/cni_safety_test.go

@@ -382,7 +382,7 @@ func TestGuardedWriteCNIConfigPreservesCollisions(t *testing.T) {
 				}
 			default:
 				got, readErr := os.ReadFile(cniConfigPath(cfg))
-				if readErr != nil || !strings.Contains(string(got), "10.244.2.0/24") {
+				if readErr != nil || !strings.Contains(string(got), `"type": "unbounded-ipam"`) || strings.Contains(string(got), "10.244.2.0/24") {
 					t.Fatalf("safe bridge did not publish current configuration: %q err=%v", got, readErr)
 				}
 

cmd/unbounded-net-node/delegated_routes_test.go

@@ -0,0 +1,90 @@
+// Copyright (c) Microsoft Corporation.
+// SPDX-License-Identifier: Apache-2.0
+
+package main
+
+import (
+	"testing"
+
+	unboundedv1alpha3 "github.com/Azure/unbounded/api/machina/v1alpha3"
+	unboundednetv1alpha1 "github.com/Azure/unbounded/api/net/v1alpha1"
+)
+
+func TestUnmanagedLocalPoolsDoNotAttractThirdPartyTraffic(t *testing.T) {
+	state := &wireGuardState{sitePodCIDRPools: []string{"10.244.0.0/16"}}
+	for _, managed := range []bool{false, true} {
+		supernets := collectSupernets(state, nil, nil, nil, nil, managed)
+		if supernets["10.244.0.0/16"] != managed {
+			t.Fatalf("local pool routing with management=%v: %v", managed, supernets)
+		}
+	}
+}
+
+func TestUnmanagedGatewayPoolsDoNotAttractThirdPartyTraffic(t *testing.T) {
+	sites := map[string]*unboundedv1alpha3.Site{
+		"local": {Spec: unboundedv1alpha3.SiteSpec{
+			ManageCniPlugin: new(false),
+			PodCidrAssignments: []unboundednetv1alpha1.PodCidrAssignment{{
+				CidrBlocks: []string{"10.244.0.0/16", "fd00:0244::/48"},
+			}},
+		}},
+		"remote": {Spec: unboundedv1alpha3.SiteSpec{
+			NodeCidrs: []string{"192.0.2.0/24"},
+			PodCidrAssignments: []unboundednetv1alpha1.PodCidrAssignment{{
+				CidrBlocks: []string{"10.245.0.0/16"},
+			}},
+		}},
+	}
+
+	routing, err := collectSiteRoutingConfig(sites, "local", true)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	routing.gatewaySupernets = append(routing.gatewaySupernets, "fd00:244::/48")
+	state := &wireGuardState{
+		sitePodCIDRPools: []string{"10.244.0.0/16", "fd00:0244::/48"},
+		sitePodCIDRs:     routing.gatewaySupernets,
+	}
+
+	gateways := []gatewayPeerInfo{{RoutedCidrs: routing.gatewaySupernets}}
+	for _, managed := range []bool{false, true} {
+		supernets := collectSupernets(state, nil, gateways, gateways, routing.gatewaySupernets, managed)
+		for _, cidr := range []string{"10.244.0.0/16", "fd00:0244::/48", "fd00:244::/48"} {
+			if supernets[cidr] != managed {
+				t.Fatalf("gateway local pool routing with management=%v: %v", managed, supernets)
+			}
+		}
+
+		for _, cidr := range []string{"10.245.0.0/16", "192.0.2.0/24"} {
+			if !supernets[cidr] {
+				t.Fatalf("gateway lost remote route %s with management=%v: %v", cidr, managed, supernets)
+			}
+		}
+	}
+}
+
+func TestUnmanagedSupernetsHonorSkippedPeerPodCIDRs(t *testing.T) {
+	state := &wireGuardState{sitePodCIDRPools: []string{"10.244.0.0/16"}}
+	peers := []meshPeerInfo{
+		{PodCIDRs: []string{"10.244.1.0/24"}, SkipPodCIDRRoutes: true},
+		{PodCIDRs: []string{"10.245.1.0/24"}},
+	}
+	gateways := []gatewayPeerInfo{
+		{PodCIDRs: []string{"10.244.2.0/24"}, SkipPodCIDRRoutes: true},
+		{PodCIDRs: []string{"10.246.1.0/24"}},
+	}
+
+	supernets := collectSupernets(state, peers, gateways, gateways, nil, false)
+	for _, cidr := range []string{"10.244.0.0/16", "10.244.1.0/24", "10.244.2.0/24"} {
+		if supernets[cidr] {
+			t.Fatalf("unmanaged local route %s was added: %v", cidr, supernets)
+		}
+	}
+
+	for _, cidr := range []string{"10.245.1.0/24", "10.246.1.0/24"} {
+		if !supernets[cidr] {
+			t.Fatalf("non-skipped peer route %s was lost: %v", cidr, supernets)
+		}
+	}
+}