Cilium agent's race condition may lead to policy bypass for Host Firewall policy

@@ -470,6 +470,11 @@ func (e *Endpoint) IsHost() bool {
 	return e.isHost
 }
 
+// SetIsHost is a convenient method to create host endpoints for testing.
+func (ep *Endpoint) SetIsHost(isHost bool) {
+	ep.isHost = isHost
+}
+
 // closeBPFProgramChannel closes the channel that signals whether the endpoint
 // has had its BPF program compiled. If the channel is already closed, this is
 // a no-op.

@@ -35,24 +35,31 @@ func (mgr *endpointManager) startNodeLabelsObserver(old map[string]string) {
 			return
 		}
 
-		mgr.updateHostEndpointLabels(old, ln.Labels)
-		old = ln.Labels
+		if mgr.updateHostEndpointLabels(old, ln.Labels) {
+			// Endpoint's label update logic rejects a request if any of the old labels are
+			// not present in the endpoint manager's state. So, overwrite old labels only if
+			// the update is successful to avoid node labels being outdated indefinitely (GH-29649).
+			old = ln.Labels
+		}
+
 	}, func(error) { /* Executed only when we are shutting down */ })
 }
 
-func (mgr *endpointManager) updateHostEndpointLabels(oldNodeLabels, newNodeLabels map[string]string) {
+// updateHostEndpointLabels updates the local node labels in the endpoint manager.
+// Returns true if the update is successful.
+func (mgr *endpointManager) updateHostEndpointLabels(oldNodeLabels, newNodeLabels map[string]string) bool {
 	nodeEP := mgr.GetHostEndpoint()
 	if nodeEP == nil {
 		log.Error("Host endpoint not found")
-		return
+		return false
 	}
 
-	err := nodeEP.UpdateLabelsFrom(oldNodeLabels, newNodeLabels, labels.LabelSourceK8s)
-	if err != nil {
+	if err := nodeEP.UpdateLabelsFrom(oldNodeLabels, newNodeLabels, labels.LabelSourceK8s); err != nil {
 		// An error can only occur if either the endpoint is terminating, or the
 		// old labels are not found. Both are impossible, hence there's no point
 		// in retrying.
 		log.WithError(err).Error("Unable to update host endpoint labels")
-		return
+		return false
 	}
+	return true
 }

@@ -21,8 +21,11 @@ import (
 	"github.com/cilium/cilium/pkg/endpoint/regeneration"
 	"github.com/cilium/cilium/pkg/fqdn/restore"
 	"github.com/cilium/cilium/pkg/hive/cell"
+	"github.com/cilium/cilium/pkg/labelsfilter"
 	"github.com/cilium/cilium/pkg/lock"
 	monitorAPI "github.com/cilium/cilium/pkg/monitor/api"
+	"github.com/cilium/cilium/pkg/node"
+	"github.com/cilium/cilium/pkg/node/types"
 	"github.com/cilium/cilium/pkg/option"
 	"github.com/cilium/cilium/pkg/policy"
 	"github.com/cilium/cilium/pkg/revert"
@@ -967,3 +970,32 @@ func (s *EndpointManagerSuite) TestWaitForEndpointsAtPolicyRev(c *C) {
 		tt.postTestRun()
 	}
 }
+
+func (s *EndpointManagerSuite) TestMissingNodeLabelsUpdate(c *C) {
+	// Initialize label filter config.
+	labelsfilter.ParseLabelPrefixCfg(nil, "")
+	mgr := New(&dummyEpSyncher{}, nil, nil)
+	hostEPID := uint16(17)
+
+	// Initialize the local node watcher before the host endpoint is created.
+	// These labels are not propagated to the endpoint manager.
+	mgr.localNodeStore = node.NewTestLocalNodeStore(node.LocalNode{Node: types.Node{}})
+	mgr.startNodeLabelsObserver(nil)
+	mgr.localNodeStore.Update(func(ln *node.LocalNode) { ln.Labels = map[string]string{"k1": "v1"} })
+	_, ok := mgr.endpoints[hostEPID]
+	c.Assert(ok, checker.Equals, false)
+
+	// Create host endpoint and expose it in the endpoint manager.
+	ep := endpoint.NewEndpointWithState(s, s, testipcache.NewMockIPCache(), &endpoint.FakeEndpointProxy{}, testidentity.NewMockIdentityAllocator(nil), 1, endpoint.StateReady)
+	ep.SetIsHost(true)
+	ep.ID = hostEPID
+	c.Assert(mgr.expose(ep), IsNil)
+
+	// Update node labels and verify that the node labels are updated correctly even if the old
+	// labels {k1=v1} are not present in the endpoint manager's state.
+	mgr.localNodeStore.Update(func(ln *node.LocalNode) { ln.Labels = map[string]string{"k2": "v2"} })
+	hostEP, ok := mgr.endpoints[hostEPID]
+	c.Assert(ok, checker.Equals, true)
+	got := hostEP.OpLabels.IdentityLabels().K8sStringMap()
+	c.Assert(map[string]string{"k2": "v2"}, checker.DeepEquals, got)
+}

@@ -506,6 +506,11 @@ func (e *Endpoint) IsHost() bool {
 	return e.isHost
 }
 
+// SetIsHost is a convenient method to create host endpoints for testing.
+func (ep *Endpoint) SetIsHost(isHost bool) {
+	ep.isHost = isHost
+}
+
 // closeBPFProgramChannel closes the channel that signals whether the endpoint
 // has had its BPF program compiled. If the channel is already closed, this is
 // a no-op.

@@ -35,24 +35,31 @@ func (mgr *endpointManager) startNodeLabelsObserver(old map[string]string) {
 			return
 		}
 
-		mgr.updateHostEndpointLabels(old, ln.Labels)
-		old = ln.Labels
+		if mgr.updateHostEndpointLabels(old, ln.Labels) {
+			// Endpoint's label update logic rejects a request if any of the old labels are
+			// not present in the endpoint manager's state. So, overwrite old labels only if
+			// the update is successful to avoid node labels being outdated indefinitely (GH-29649).
+			old = ln.Labels
+		}
+
 	}, func(error) { /* Executed only when we are shutting down */ })
 }
 
-func (mgr *endpointManager) updateHostEndpointLabels(oldNodeLabels, newNodeLabels map[string]string) {
+// updateHostEndpointLabels updates the local node labels in the endpoint manager.
+// Returns true if the update is successful.
+func (mgr *endpointManager) updateHostEndpointLabels(oldNodeLabels, newNodeLabels map[string]string) bool {
 	nodeEP := mgr.GetHostEndpoint()
 	if nodeEP == nil {
 		log.Error("Host endpoint not found")
-		return
+		return false
 	}
 
-	err := nodeEP.UpdateLabelsFrom(oldNodeLabels, newNodeLabels, labels.LabelSourceK8s)
-	if err != nil {
+	if err := nodeEP.UpdateLabelsFrom(oldNodeLabels, newNodeLabels, labels.LabelSourceK8s); err != nil {
 		// An error can only occur if either the endpoint is terminating, or the
 		// old labels are not found. Both are impossible, hence there's no point
 		// in retrying.
 		log.WithError(err).Error("Unable to update host endpoint labels")
-		return
+		return false
 	}
+	return true
 }

@@ -19,7 +19,10 @@ import (
 	"github.com/cilium/cilium/pkg/endpoint"
 	endpointid "github.com/cilium/cilium/pkg/endpoint/id"
 	"github.com/cilium/cilium/pkg/fqdn/restore"
+	"github.com/cilium/cilium/pkg/labelsfilter"
 	monitorAPI "github.com/cilium/cilium/pkg/monitor/api"
+	"github.com/cilium/cilium/pkg/node"
+	"github.com/cilium/cilium/pkg/node/types"
 	"github.com/cilium/cilium/pkg/option"
 	"github.com/cilium/cilium/pkg/policy"
 	testidentity "github.com/cilium/cilium/pkg/testutils/identity"
@@ -958,3 +961,33 @@ func TestWaitForEndpointsAtPolicyRev(t *testing.T) {
 		tt.postTestRun()
 	}
 }
+
+func TestMissingNodeLabelsUpdate(t *testing.T) {
+	// Initialize label filter config.
+	labelsfilter.ParseLabelPrefixCfg(nil, nil, "")
+	s := setupEndpointManagerSuite(t)
+	mgr := New(&dummyEpSyncher{}, nil, nil)
+	hostEPID := uint16(17)
+
+	// Initialize the local node watcher before the host endpoint is created.
+	// These labels are not propagated to the endpoint manager.
+	mgr.localNodeStore = node.NewTestLocalNodeStore(node.LocalNode{Node: types.Node{}})
+	mgr.startNodeLabelsObserver(nil)
+	mgr.localNodeStore.Update(func(ln *node.LocalNode) { ln.Labels = map[string]string{"k1": "v1"} })
+	_, ok := mgr.endpoints[hostEPID]
+	require.EqualValues(t, ok, false)
+
+	// Create host endpoint and expose it in the endpoint manager.
+	ep := endpoint.NewTestEndpointWithState(t, s, s, testipcache.NewMockIPCache(), &endpoint.FakeEndpointProxy{}, testidentity.NewMockIdentityAllocator(nil), 1, endpoint.StateReady)
+	ep.SetIsHost(true)
+	ep.ID = hostEPID
+	require.Nil(t, mgr.expose(ep))
+
+	// Update node labels and verify that the node labels are updated correctly even if the old
+	// labels {k1=v1} are not present in the endpoint manager's state.
+	mgr.localNodeStore.Update(func(ln *node.LocalNode) { ln.Labels = map[string]string{"k2": "v2"} })
+	hostEP, ok := mgr.endpoints[hostEPID]
+	require.EqualValues(t, ok, true)
+	got := hostEP.OpLabels.IdentityLabels().K8sStringMap()
+	require.EqualValues(t, map[string]string{"k2": "v2"}, got)
+}

@@ -410,6 +410,11 @@ func (e *Endpoint) IsHost() bool {
 	return e.isHost
 }
 
+// SetIsHost is a convinient method to create host endpoints for testing.
+func (ep *Endpoint) SetIsHost(isHost bool) {
+	ep.isHost = isHost
+}
+
 // closeBPFProgramChannel closes the channel that signals whether the endpoint
 // has had its BPF program compiled. If the channel is already closed, this is
 // a no-op.

@@ -4,16 +4,18 @@
 package endpointmanager
 
 import (
-	"context"
-
 	"github.com/cilium/cilium/pkg/endpoint"
 	slim_corev1 "github.com/cilium/cilium/pkg/k8s/slim/k8s/api/core/v1"
 	"github.com/cilium/cilium/pkg/labels"
-	"github.com/cilium/cilium/pkg/labelsfilter"
 	"github.com/cilium/cilium/pkg/lock"
 	"github.com/cilium/cilium/pkg/node"
 )
 
+// isFirstNodeUpdate tracks the first node update to the endpoint manager.
+// This is used to resolve a race condition during agent startup where the
+// k8s node label updates are rejected indefinitely by the host endpoint (GH-29649).
+var isFirstNodeUpdate = true
+
 // GetHostEndpoint returns the host endpoint.
 func (mgr *endpointManager) GetHostEndpoint() *endpoint.Endpoint {
 	mgr.mutex.RLock()
@@ -36,20 +38,7 @@ func (mgr *endpointManager) HostEndpointExists() bool {
 // This adheres to the subscriber.NodeHandler interface.
 func (mgr *endpointManager) OnAddNode(newNode *slim_corev1.Node,
 	swg *lock.StoppableWaitGroup) error {
-
-	node.SetLabels(newNode.GetLabels())
-
-	nodeEP := mgr.GetHostEndpoint()
-	if nodeEP == nil {
-		// if host endpoint does not exist yet, labels will be set when it'll be created.
-		return nil
-	}
-
-	newLabels := labels.Map2Labels(newNode.GetLabels(), labels.LabelSourceK8s)
-	newIdtyLabels, _ := labelsfilter.Filter(newLabels)
-	nodeEP.UpdateLabels(context.TODO(), labels.LabelSourceAny, newIdtyLabels, nil, false)
-
-	return nil
+	return mgr.OnUpdateNode(nil, newNode, swg)
 }
 
 // OnUpdateNode implements the endpointManager's logic for reacting to updated
@@ -58,21 +47,31 @@ func (mgr *endpointManager) OnAddNode(newNode *slim_corev1.Node,
 func (mgr *endpointManager) OnUpdateNode(oldNode, newNode *slim_corev1.Node,
 	swg *lock.StoppableWaitGroup) error {
 
-	oldNodeLabels := oldNode.GetLabels()
+	var oldNodeLabels map[string]string
+	// Endpoint's label update logic rejects a request if any of the old labels are
+	// not present in the endpoint manager's state. Overwrite the old labels to nil
+	// for the first node update to avoid node labels being outdated indefinitely (GH-29649).
+	if oldNode == nil || isFirstNodeUpdate {
+		oldNodeLabels = make(map[string]string)
+	} else {
+		oldNodeLabels = oldNode.GetLabels()
+	}
 	newNodeLabels := newNode.GetLabels()
+	// Set the labels early so host endpoint is created with latest labels.
+	node.SetLabels(newNodeLabels)
 
 	nodeEP := mgr.GetHostEndpoint()
 	if nodeEP == nil {
-		log.Error("Host endpoint not found")
+		log.Debug("Host endpoint not found")
 		return nil
 	}
 
-	node.SetLabels(newNodeLabels)
-
-	err := nodeEP.UpdateLabelsFrom(oldNodeLabels, newNodeLabels, labels.LabelSourceK8s)
-	if err != nil {
+	if err := nodeEP.UpdateLabelsFrom(oldNodeLabels, newNodeLabels, labels.LabelSourceK8s); err != nil {
+		log.WithError(err).Error("Unable to update host endpoint labels")
 		return err
 	}
+	// Set isFirstNodeUpdate to false only on successful update.
+	isFirstNodeUpdate = false
 
 	return nil
 }

@@ -20,8 +20,12 @@ import (
 	endpointid "github.com/cilium/cilium/pkg/endpoint/id"
 	"github.com/cilium/cilium/pkg/endpoint/regeneration"
 	"github.com/cilium/cilium/pkg/fqdn/restore"
+	slim_corev1 "github.com/cilium/cilium/pkg/k8s/slim/k8s/api/core/v1"
+	slim_metav1 "github.com/cilium/cilium/pkg/k8s/slim/k8s/apis/meta/v1"
+	"github.com/cilium/cilium/pkg/labelsfilter"
 	"github.com/cilium/cilium/pkg/lock"
 	monitorAPI "github.com/cilium/cilium/pkg/monitor/api"
+	"github.com/cilium/cilium/pkg/node"
 	"github.com/cilium/cilium/pkg/option"
 	"github.com/cilium/cilium/pkg/policy"
 	"github.com/cilium/cilium/pkg/revert"
@@ -815,3 +819,54 @@ func (s *EndpointManagerSuite) TestWaitForEndpointsAtPolicyRev(c *C) {
 		tt.postTestRun()
 	}
 }
+
+func (s *EndpointManagerSuite) TestMissingNodeLabelsUpdate(c *C) {
+	node.SetTestLocalNodeStore()
+	defer node.UnsetTestLocalNodeStore()
+	// Initialize label filter config.
+	labelsfilter.ParseLabelPrefixCfg(nil, "")
+	mgr := New(&dummyEpSyncher{})
+	hostEPID := uint16(17)
+	hostEP := endpoint.NewEndpointWithState(s, s, testipcache.NewMockIPCache(), &endpoint.FakeEndpointProxy{}, testidentity.NewMockIdentityAllocator(nil), 1, endpoint.StateReady)
+	hostEP.SetIsHost(true)
+	hostEP.ID = hostEPID
+
+	// Helper function to create test node object.
+	newTestNode := func(labels map[string]string) *slim_corev1.Node {
+		return &slim_corev1.Node{
+			ObjectMeta: slim_metav1.ObjectMeta{
+				Labels: labels,
+			},
+		}
+	}
+
+	// Validate that labels for host endpoint are correctly set.
+	validateHostEPLabels := func(wantLabels map[string]string) {
+		currHostIP, ok := mgr.endpoints[hostEPID]
+		c.Assert(ok, Equals, true)
+		c.Assert(currHostIP.OpLabels.IdentityLabels().K8sStringMap(), checker.DeepEquals, wantLabels)
+	}
+
+	oldNode := newTestNode(make(map[string]string))
+	newNode := newTestNode(map[string]string{"k1": "v1"})
+
+	// Host endpoint is not created in endpoint manager so the node update
+	// is not consumed by endpoint manager.
+	c.Assert(mgr.OnUpdateNode(oldNode, newNode, lock.NewStoppableWaitGroup()), IsNil)
+	_, ok := mgr.endpoints[hostEPID]
+	c.Assert(ok, Equals, false)
+
+	// Now expose host endpoint.
+	mgr.expose(hostEP)
+
+	// Even though endpoint manager does not have k1=v1, the request is still accepted.
+	oldNode = newTestNode(map[string]string{"k1": "v1"})
+	newNode = newTestNode(map[string]string{"k2": "v2"})
+	c.Assert(mgr.OnUpdateNode(oldNode, newNode, lock.NewStoppableWaitGroup()), IsNil)
+	validateHostEPLabels(map[string]string{"k2": "v2"})
+
+	// k3=v3 is not present in the endpoint manager so an error is not returned.
+	oldNode = newTestNode(map[string]string{"k3": "v3"})
+	newNode = newTestNode(map[string]string{"k4": "v4"})
+	c.Assert(mgr.OnUpdateNode(oldNode, newNode, lock.NewStoppableWaitGroup()), NotNil)
+}