From 02e90171dfbab2f7e758f79d7b3e55bb42be5e32 Mon Sep 17 00:00:00 2001 From: Gaurav Chaudhary Date: Fri, 4 Sep 2026 22:16:27 +0530 Subject: [PATCH] fix: remap Service egress ports to enforcement ports for NetworkPolicy (#943) eBPF observes the Service port before kube-proxy DNAT, but CNIs enforce on the pod targetPort. Resolve numeric, omitted, and named targetPorts (via EndpointSlice with Endpoints fallback) into NetworkNeighbor.Ports when serializing egress neighbors so generated policies open the port the CNI actually matches. Companion helm-charts PR required for EndpointSlice RBAC in production. Signed-off-by: Gaurav Chaudhary --- .../v1/container_data.go | 23 +- .../v1/container_data_test.go | 381 ++++++++++++++++++ .../v1/event_reporting.go | 14 +- .../v1/network_helpers_service_ports.go | 257 ++++++++++++ 4 files changed, 664 insertions(+), 11 deletions(-) create mode 100644 pkg/containerprofilemanager/v1/container_data_test.go create mode 100644 pkg/containerprofilemanager/v1/network_helpers_service_ports.go diff --git a/pkg/containerprofilemanager/v1/container_data.go b/pkg/containerprofilemanager/v1/container_data.go index 7f07ad5c2..ebfae38c7 100644 --- a/pkg/containerprofilemanager/v1/container_data.go +++ b/pkg/containerprofilemanager/v1/container_data.go @@ -6,12 +6,12 @@ import ( mapset "github.com/deckarep/golang-set/v2" "github.com/kubescape/go-logger" "github.com/kubescape/go-logger/helpers" + "github.com/kubescape/k8s-interface/k8sinterface" "github.com/kubescape/node-agent/pkg/dnsmanager" "github.com/kubescape/node-agent/pkg/k8sclient" "github.com/kubescape/node-agent/pkg/utils" "github.com/kubescape/storage/pkg/apis/softwarecomposition/v1beta1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "k8s.io/utils/ptr" ) // emptyEvents clears all event data @@ -202,12 +202,8 @@ func (cd *containerData) getEgressNetworkNeighbors(containerID string, namespace func (cd *containerData) createNetworkNeighbor(containerID string, networkEvent NetworkEvent, namespace string, k8sClient k8sclient.K8sClientInterface, dnsResolverClient dnsmanager.DNSResolver) *v1beta1.NetworkNeighbor { var neighborEntry v1beta1.NetworkNeighbor - portIdentifier := generatePortIdentifierFromEvent(networkEvent) - neighborEntry.Ports = []v1beta1.NetworkPort{{ - Protocol: v1beta1.Protocol(networkEvent.Protocol), - Port: ptr.To(int32(networkEvent.Port)), - Name: portIdentifier, - }} + enforcementPorts := []uint16{networkEvent.Port} + var serviceWorkload k8sinterface.IWorkload if networkEvent.Destination.Kind == EndpointKindPod { // For Pods, we need to remove the default labels @@ -230,6 +226,7 @@ func (cd *containerData) createNetworkNeighbor(containerID string, networkEvent helpers.String("service name", networkEvent.Destination.Name)) return nil } + serviceWorkload = svc var selector map[string]string if svc.GetName() == "kubernetes" && svc.GetNamespace() == "default" { @@ -270,6 +267,18 @@ func (cd *containerData) createNetworkNeighbor(containerID string, networkEvent } } + if networkEvent.Destination.Kind == EndpointKindService && serviceWorkload != nil && k8sClient != nil { + enforcementPorts = resolveServiceEnforcementPorts( + k8sClient, + networkEvent.Destination.Namespace, + networkEvent.Destination.Name, + serviceWorkload, + networkEvent.Port, + networkEvent.Protocol, + ) + } + neighborEntry.Ports = buildNetworkPorts(networkEvent.Protocol, enforcementPorts) + neighborEntry.Type = InternalTrafficType if neighborEntry.NamespaceSelector == nil && neighborEntry.PodSelector == nil { neighborEntry.Type = ExternalTrafficType diff --git a/pkg/containerprofilemanager/v1/container_data_test.go b/pkg/containerprofilemanager/v1/container_data_test.go new file mode 100644 index 000000000..39e92260c --- /dev/null +++ b/pkg/containerprofilemanager/v1/container_data_test.go @@ -0,0 +1,381 @@ +package containerprofilemanager + +import ( + "fmt" + "testing" + + helpersv1 "github.com/kubescape/k8s-interface/instanceidhandler/v1/helpers" + "github.com/kubescape/k8s-interface/k8sinterface" + "github.com/kubescape/k8s-interface/workloadinterface" + "github.com/kubescape/node-agent/pkg/k8sclient" + "github.com/kubescape/node-agent/pkg/utils" + "github.com/kubescape/storage/pkg/apis/softwarecomposition" + "github.com/kubescape/storage/pkg/apis/softwarecomposition/networkpolicy/v2" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + corev1 "k8s.io/api/core/v1" + discoveryv1 "k8s.io/api/discovery/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/client-go/dynamic" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/kubernetes/fake" + "k8s.io/utils/ptr" +) + +type servicePortTestClient struct { + service k8sinterface.IWorkload + endpoints k8sinterface.IWorkload + kubeClient *fake.Clientset + getErr error +} + +var _ k8sclient.K8sClientInterface = (*servicePortTestClient)(nil) + +func (c *servicePortTestClient) GetWorkload(_, kind, _ string) (k8sinterface.IWorkload, error) { + if c.getErr != nil { + return nil, c.getErr + } + switch kind { + case "Service": + return c.service, nil + case "Endpoints": + return c.endpoints, nil + default: + return nil, fmt.Errorf("unsupported kind %q", kind) + } +} + +func (c *servicePortTestClient) CalculateWorkloadParentRecursive(workload k8sinterface.IWorkload) (string, string, error) { + return workload.GetKind(), workload.GetName(), nil +} + +func (c *servicePortTestClient) GetKubernetesClient() kubernetes.Interface { + return c.kubeClient +} + +func (c *servicePortTestClient) GetDynamicClient() dynamic.Interface { + return nil +} + +func newServiceWorkload(name string, selector map[string]interface{}, ports ...map[string]interface{}) k8sinterface.IWorkload { + portEntries := make([]interface{}, 0, len(ports)) + for _, port := range ports { + portEntries = append(portEntries, port) + } + return workloadinterface.NewWorkloadObj(map[string]interface{}{ + "apiVersion": "v1", + "kind": "Service", + "metadata": map[string]interface{}{ + "name": name, + "namespace": "default", + }, + "spec": map[string]interface{}{ + "selector": selector, + "ports": portEntries, + }, + }) +} + +func newEndpointsWorkload(name string, ports ...map[string]interface{}) k8sinterface.IWorkload { + portEntries := make([]interface{}, 0, len(ports)) + for _, port := range ports { + portEntries = append(portEntries, port) + } + return workloadinterface.NewWorkloadObj(map[string]interface{}{ + "apiVersion": "v1", + "kind": "Endpoints", + "metadata": map[string]interface{}{ + "name": name, + "namespace": "default", + }, + "subsets": []interface{}{ + map[string]interface{}{ + "ports": portEntries, + }, + }, + }) +} + +func newEndpointSlice(name, serviceName string, ports ...discoveryv1.EndpointPort) *discoveryv1.EndpointSlice { + return &discoveryv1.EndpointSlice{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Namespace: "default", + Labels: map[string]string{ + discoveryv1.LabelServiceName: serviceName, + }, + }, + Ports: ports, + } +} + +func serviceNetworkEvent(port uint16, protocol string) NetworkEvent { + return NetworkEvent{ + Port: port, + Protocol: protocol, + PktType: utils.OutgoingPktType, + Destination: Destination{ + Kind: EndpointKindService, + Namespace: "default", + Name: "api", + }, + } +} + +func TestCreateNetworkNeighbor_ServiceTargetPortMatrix(t *testing.T) { + tests := []struct { + name string + service k8sinterface.IWorkload + endpoints k8sinterface.IWorkload + endpointSlice []*discoveryv1.EndpointSlice + event NetworkEvent + wantPorts []int32 + }{ + { + name: "numeric remap", + service: newServiceWorkload("api", map[string]interface{}{"app": "api"}, map[string]interface{}{ + "port": 80, "targetPort": 8080, "protocol": "TCP", + }), + event: serviceNetworkEvent(80, "tcp"), + wantPorts: []int32{8080}, + }, + { + name: "unchanged when port equals targetPort", + service: newServiceWorkload("api", map[string]interface{}{"app": "api"}, map[string]interface{}{ + "port": 8080, "targetPort": 8080, "protocol": "TCP", + }), + event: serviceNetworkEvent(8080, "tcp"), + wantPorts: []int32{8080}, + }, + { + name: "omitted targetPort defaults to service port", + service: newServiceWorkload("api", map[string]interface{}{"app": "api"}, map[string]interface{}{ + "port": 80, "protocol": "TCP", + }), + event: serviceNetworkEvent(80, "tcp"), + wantPorts: []int32{80}, + }, + { + name: "udp remap", + service: newServiceWorkload("api", map[string]interface{}{"app": "api"}, map[string]interface{}{ + "port": 53, "targetPort": 5353, "protocol": "UDP", + }), + event: serviceNetworkEvent(53, "udp"), + wantPorts: []int32{5353}, + }, + { + name: "multi-port service selects matching service port", + service: newServiceWorkload("api", map[string]interface{}{"app": "api"}, + map[string]interface{}{"port": 80, "targetPort": 8080, "protocol": "TCP"}, + map[string]interface{}{"port": 443, "targetPort": 8443, "protocol": "TCP"}, + ), + event: serviceNetworkEvent(443, "tcp"), + wantPorts: []int32{8443}, + }, + { + name: "protocol mismatch keeps observed port", + service: newServiceWorkload("api", map[string]interface{}{"app": "api"}, map[string]interface{}{ + "port": 80, "targetPort": 8080, "protocol": "TCP", + }), + event: serviceNetworkEvent(80, "udp"), + wantPorts: []int32{80}, + }, + { + name: "unknown observed port falls back", + service: newServiceWorkload("api", map[string]interface{}{"app": "api"}, map[string]interface{}{ + "port": 80, "targetPort": 8080, "protocol": "TCP", + }), + event: serviceNetworkEvent(9999, "tcp"), + wantPorts: []int32{9999}, + }, + { + name: "malformed service falls back safely", + service: workloadinterface.NewWorkloadObj(map[string]interface{}{ + "apiVersion": "v1", + "kind": "Service", + "metadata": map[string]interface{}{"name": "api", "namespace": "default"}, + "spec": map[string]interface{}{ + "selector": map[string]interface{}{"app": "api"}, + "ports": "invalid", + }, + }), + event: serviceNetworkEvent(80, "tcp"), + wantPorts: []int32{80}, + }, + { + name: "named targetPort resolves via endpointslice on service port name", + service: newServiceWorkload("api", map[string]interface{}{"app": "api"}, map[string]interface{}{ + "name": "web", "port": 80, "targetPort": "http", "protocol": "TCP", + }), + endpointSlice: []*discoveryv1.EndpointSlice{ + newEndpointSlice("api-a", "api", discoveryv1.EndpointPort{ + Name: ptr.To("web"), + Port: ptr.To(int32(8080)), + Protocol: ptr.To(corev1.ProtocolTCP), + }), + }, + event: serviceNetworkEvent(80, "tcp"), + wantPorts: []int32{8080}, + }, + { + name: "heterogeneous named targetPort collects all endpoint ports", + service: newServiceWorkload("api", map[string]interface{}{"app": "api"}, map[string]interface{}{ + "name": "web", "port": 80, "targetPort": "http", "protocol": "TCP", + }), + endpointSlice: []*discoveryv1.EndpointSlice{ + newEndpointSlice("api-a", "api", discoveryv1.EndpointPort{ + Name: ptr.To("web"), + Port: ptr.To(int32(8080)), + Protocol: ptr.To(corev1.ProtocolTCP), + }), + newEndpointSlice("api-b", "api", discoveryv1.EndpointPort{ + Name: ptr.To("web"), + Port: ptr.To(int32(9090)), + Protocol: ptr.To(corev1.ProtocolTCP), + }), + }, + event: serviceNetworkEvent(80, "tcp"), + wantPorts: []int32{8080, 9090}, + }, + { + name: "endpoints fallback when no endpointslice", + service: newServiceWorkload("api", map[string]interface{}{"app": "api"}, map[string]interface{}{ + "name": "web", "port": 80, "targetPort": "http", "protocol": "TCP", + }), + endpoints: newEndpointsWorkload("api", map[string]interface{}{ + "name": "web", "port": 8080, "protocol": "TCP", + }), + event: serviceNetworkEvent(80, "tcp"), + wantPorts: []int32{8080}, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + objects := make([]runtime.Object, 0, len(tc.endpointSlice)) + for _, slice := range tc.endpointSlice { + objects = append(objects, slice) + } + client := &servicePortTestClient{ + service: tc.service, + endpoints: tc.endpoints, + kubeClient: fake.NewClientset(objects...), + } + + cd := &containerData{} + neighbor := cd.createNetworkNeighbor("", tc.event, "default", client, nil) + require.NotNil(t, neighbor) + require.Equal(t, map[string]string{"app": "api"}, neighbor.PodSelector.MatchLabels) + require.Equal(t, tc.wantPorts, networkPortValues(neighbor.Ports)) + + if tc.name == "numeric remap" { + require.Equal(t, int32(8080), *neighbor.Ports[0].Port) + require.NotEqual(t, int32(80), *neighbor.Ports[0].Port) + require.Equal(t, "tcp-8080", neighbor.Ports[0].Name) + } + }) + } +} + +func TestCreateNetworkNeighbor_NonServiceDestinationsUnchanged(t *testing.T) { + cd := &containerData{} + + podEvent := NetworkEvent{ + Port: 8080, + Protocol: "tcp", + PktType: utils.OutgoingPktType, + Destination: Destination{ + Kind: EndpointKindPod, + Namespace: "default", + Name: "web", + }, + } + podEvent.SetDestinationPodLabels(map[string]string{"app": "web"}) + podNeighbor := cd.createNetworkNeighbor("", podEvent, "default", nil, nil) + require.NotNil(t, podNeighbor) + require.Equal(t, []int32{8080}, networkPortValues(podNeighbor.Ports)) + + rawEvent := NetworkEvent{ + Port: 443, + Protocol: "tcp", + PktType: utils.OutgoingPktType, + Destination: Destination{ + IPAddress: "93.184.216.34", + }, + } + rawNeighbor := cd.createNetworkNeighbor("", rawEvent, "default", nil, nil) + require.NotNil(t, rawNeighbor) + require.Equal(t, []int32{443}, networkPortValues(rawNeighbor.Ports)) +} + +func TestGenerateNetworkPolicy_ServiceTargetPortRoundTrip(t *testing.T) { + service := newServiceWorkload("api", map[string]interface{}{"app.kubernetes.io/name": "api"}, map[string]interface{}{ + "port": 80, "targetPort": 8080, "protocol": "TCP", + }) + client := &servicePortTestClient{ + service: service, + kubeClient: fake.NewClientset(), + } + + cd := &containerData{} + event := serviceNetworkEvent(80, "tcp") + neighbor := cd.createNetworkNeighbor("", event, "default", client, nil) + require.NotNil(t, neighbor) + + egressPorts := make([]softwarecomposition.NetworkPort, 0, len(neighbor.Ports)) + for _, port := range neighbor.Ports { + egressPorts = append(egressPorts, softwarecomposition.NetworkPort{ + Protocol: softwarecomposition.Protocol(port.Protocol), + Port: port.Port, + Name: port.Name, + }) + } + + cp := &softwarecomposition.ContainerProfile{ + ObjectMeta: metav1.ObjectMeta{ + Name: "deployment-client", + Namespace: "default", + Labels: map[string]string{ + helpersv1.RelatedKindMetadataKey: "Deployment", + helpersv1.RelatedNameMetadataKey: "client", + }, + Annotations: map[string]string{ + helpersv1.StatusMetadataKey: helpersv1.Completed, + }, + }, + Spec: softwarecomposition.ContainerProfileSpec{ + LabelSelector: metav1.LabelSelector{ + MatchLabels: map[string]string{"app": "client"}, + }, + Egress: []softwarecomposition.NetworkNeighbor{{ + PodSelector: neighbor.PodSelector, + Ports: egressPorts, + Type: softwarecomposition.CommunicationType(neighbor.Type), + }}, + }, + } + + gnp, err := networkpolicy.GenerateNetworkPolicy(cp, softwarecomposition.NewKnownServersFinderImpl(nil), metav1.Now()) + require.NoError(t, err) + require.NotEmpty(t, gnp.Spec.Spec.Egress) + + var found8080 bool + for _, rule := range gnp.Spec.Spec.Egress { + for _, port := range rule.Ports { + if port.Port != nil && *port.Port == 8080 { + found8080 = true + } + if port.Port != nil { + assert.NotEqual(t, int32(80), *port.Port) + } + } + } + assert.True(t, found8080) +} + +func TestResolveServiceEnforcementPorts_Unit(t *testing.T) { + t.Run("dedupe and sort", func(t *testing.T) { + require.Equal(t, []uint16{8080, 9090}, dedupeSortPorts([]uint16{9090, 8080, 9090})) + }) +} diff --git a/pkg/containerprofilemanager/v1/event_reporting.go b/pkg/containerprofilemanager/v1/event_reporting.go index a0b07c5bc..24839821a 100644 --- a/pkg/containerprofilemanager/v1/event_reporting.go +++ b/pkg/containerprofilemanager/v1/event_reporting.go @@ -71,10 +71,11 @@ var maxServiceSelectorEstimate = func() int { // caller, but re-shaping that string into map[string]string costs real additional bytes (Go // map bucket overhead), so this measures that wrapper delta exactly from the same data // filterLabels/GetDestinationPodLabels would produce, rather than assuming it's zero or -// guessing a label count. Ports and NamespaceSelector are similarly computed exactly from -// fields already on the event (Port/Protocol, and the destination namespace compared against -// the container's own), not estimated, since nothing about their content is deferred to -// serialization. +// guessing a label count. NamespaceSelector is computed exactly from fields already on the +// event. Port/Protocol on Pod and raw branches are also exact from the event; on the Service +// branch serialization may remap observed socket ports to enforcement (target/endpoint) +// ports and may emit multiple NetworkPort entries, so the Service case budgets a +// conservative upper bound instead. func networkNeighborIncrement(data *containerData, networkEvent NetworkEvent) int { est := neighborFixedOverhead + size.Of([]v1beta1.NetworkPort{{ Name: generatePortIdentifierFromEvent(networkEvent), @@ -93,6 +94,11 @@ func networkNeighborIncrement(data *containerData, networkEvent NetworkEvent) in switch networkEvent.Destination.Kind { case EndpointKindService: est += maxServiceSelectorEstimate + est += size.Of(v1beta1.NetworkPort{ + Name: generatePortIdentifier(networkEvent.Protocol, 65535), + Protocol: v1beta1.Protocol(networkEvent.Protocol), + Port: ptr.To(int32(65535)), + }) case EndpointKindPod: // The label bytes are already on the meter via Destination.PodLabels; only charge the // extra cost of wrapping them into a LabelSelector's map, and never a negative one. diff --git a/pkg/containerprofilemanager/v1/network_helpers_service_ports.go b/pkg/containerprofilemanager/v1/network_helpers_service_ports.go new file mode 100644 index 000000000..6010ec88d --- /dev/null +++ b/pkg/containerprofilemanager/v1/network_helpers_service_ports.go @@ -0,0 +1,257 @@ +package containerprofilemanager + +import ( + "context" + "sort" + "strconv" + "strings" + + "github.com/kubescape/go-logger" + "github.com/kubescape/go-logger/helpers" + "github.com/kubescape/k8s-interface/k8sinterface" + "github.com/kubescape/node-agent/pkg/k8sclient" + "github.com/kubescape/storage/pkg/apis/softwarecomposition/v1beta1" + corev1 "k8s.io/api/core/v1" + discoveryv1 "k8s.io/api/discovery/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/util/intstr" + "k8s.io/utils/ptr" +) + +func buildNetworkPorts(protocol string, ports []uint16) []v1beta1.NetworkPort { + networkPorts := make([]v1beta1.NetworkPort, 0, len(ports)) + for _, port := range ports { + networkPorts = append(networkPorts, v1beta1.NetworkPort{ + Protocol: v1beta1.Protocol(protocol), + Port: ptr.To(int32(port)), + Name: generatePortIdentifier(protocol, int32(port)), + }) + } + return networkPorts +} + +func resolveServiceEnforcementPorts( + k8sClient k8sclient.K8sClientInterface, + namespace, serviceName string, + svc k8sinterface.IWorkload, + observedPort uint16, + protocol string, +) []uint16 { + if svc == nil { + return []uint16{observedPort} + } + + var service corev1.Service + if err := runtime.DefaultUnstructuredConverter.FromUnstructured(svc.GetObject(), &service); err != nil { + logger.L().Warning("failed to convert service for port resolution", + helpers.String("reason", err.Error()), + helpers.String("service", serviceName), + helpers.String("namespace", namespace)) + return []uint16{observedPort} + } + + normalizedProto := normalizeProtocol(protocol) + for _, sp := range service.Spec.Ports { + if sp.Port != int32(observedPort) { + continue + } + if !protocolsMatch(sp.Protocol, normalizedProto) { + continue + } + + switch sp.TargetPort.Type { + case intstr.Int: + if sp.TargetPort.IntVal > 0 { + return []uint16{uint16(sp.TargetPort.IntVal)} + } + return []uint16{uint16(sp.Port)} + case intstr.String: + if port, ok := parseNumericPortString(sp.TargetPort.StrVal); ok { + return []uint16{port} + } + if ports := resolveEndpointPortsByServicePortName(k8sClient, namespace, serviceName, sp.Name, protocol); len(ports) > 0 { + return ports + } + return []uint16{observedPort} + default: + return []uint16{uint16(sp.Port)} + } + } + + return []uint16{observedPort} +} + +func resolveEndpointPortsByServicePortName( + k8sClient k8sclient.K8sClientInterface, + namespace, serviceName, servicePortName, protocol string, +) []uint16 { + if k8sClient == nil { + return nil + } + + normalizedProto := normalizeProtocol(protocol) + if ports := collectEndpointSlicePorts(k8sClient, namespace, serviceName, servicePortName, normalizedProto); len(ports) > 0 { + return dedupeSortPorts(ports) + } + if ports := collectEndpointsPorts(k8sClient, namespace, serviceName, servicePortName, normalizedProto); len(ports) > 0 { + return dedupeSortPorts(ports) + } + return nil +} + +func collectEndpointSlicePorts( + k8sClient k8sclient.K8sClientInterface, + namespace, serviceName, servicePortName string, + normalizedProto corev1.Protocol, +) []uint16 { + client := k8sClient.GetKubernetesClient() + if client == nil { + return nil + } + + slices, err := client.DiscoveryV1().EndpointSlices(namespace).List(context.Background(), metav1.ListOptions{ + LabelSelector: discoveryv1.LabelServiceName + "=" + serviceName, + }) + if err != nil { + if apierrors.IsForbidden(err) { + logger.L().Debug("endpointslice list forbidden, falling back to endpoints", + helpers.String("service", serviceName), + helpers.String("namespace", namespace)) + } else { + logger.L().Warning("failed to list endpointslices", + helpers.String("reason", err.Error()), + helpers.String("service", serviceName), + helpers.String("namespace", namespace)) + } + return nil + } + + var ports []uint16 + for i := range slices.Items { + for _, endpointPort := range slices.Items[i].Ports { + if !endpointSlicePortMatches(endpointPort, servicePortName, normalizedProto) { + continue + } + if endpointPort.Port != nil { + ports = append(ports, uint16(*endpointPort.Port)) + } + } + } + return ports +} + +func collectEndpointsPorts( + k8sClient k8sclient.K8sClientInterface, + namespace, serviceName, servicePortName string, + normalizedProto corev1.Protocol, +) []uint16 { + endpointsObj, err := k8sClient.GetWorkload(namespace, "Endpoints", serviceName) + if err != nil { + logger.L().Debug("failed to get endpoints for port resolution", + helpers.String("reason", err.Error()), + helpers.String("service", serviceName), + helpers.String("namespace", namespace)) + return nil + } + + var endpoints corev1.Endpoints + if err := runtime.DefaultUnstructuredConverter.FromUnstructured(endpointsObj.GetObject(), &endpoints); err != nil { + logger.L().Warning("failed to convert endpoints for port resolution", + helpers.String("reason", err.Error()), + helpers.String("service", serviceName), + helpers.String("namespace", namespace)) + return nil + } + + var ports []uint16 + for _, subset := range endpoints.Subsets { + for _, endpointPort := range subset.Ports { + if !endpointPortMatches(endpointPort, servicePortName, normalizedProto) { + continue + } + ports = append(ports, uint16(endpointPort.Port)) + } + } + return ports +} + +func endpointSlicePortMatches(port discoveryv1.EndpointPort, servicePortName string, normalizedProto corev1.Protocol) bool { + if port.Name == nil || *port.Name != servicePortName { + return false + } + if port.Protocol != nil && !protocolsMatch(*port.Protocol, normalizedProto) { + return false + } + return port.Port != nil +} + +func endpointPortMatches(port corev1.EndpointPort, servicePortName string, normalizedProto corev1.Protocol) bool { + if port.Name != servicePortName { + return false + } + if port.Protocol != "" && !protocolsMatch(port.Protocol, normalizedProto) { + return false + } + return port.Port > 0 +} + +func normalizeProtocol(protocol string) corev1.Protocol { + switch strings.ToUpper(protocol) { + case string(corev1.ProtocolUDP): + return corev1.ProtocolUDP + case string(corev1.ProtocolSCTP): + return corev1.ProtocolSCTP + default: + return corev1.ProtocolTCP + } +} + +func protocolsMatch(left, right corev1.Protocol) bool { + if left == "" { + left = corev1.ProtocolTCP + } + if right == "" { + right = corev1.ProtocolTCP + } + return left == right +} + +func parseNumericPortString(value string) (uint16, bool) { + n, err := strconv.ParseUint(value, 10, 16) + if err != nil { + return 0, false + } + return uint16(n), true +} + +func dedupeSortPorts(ports []uint16) []uint16 { + if len(ports) == 0 { + return nil + } + + seen := make(map[uint16]struct{}, len(ports)) + unique := make([]uint16, 0, len(ports)) + for _, port := range ports { + if _, ok := seen[port]; ok { + continue + } + seen[port] = struct{}{} + unique = append(unique, port) + } + + sort.Slice(unique, func(i, j int) bool { return unique[i] < unique[j] }) + return unique +} + +func networkPortValues(ports []v1beta1.NetworkPort) []int32 { + values := make([]int32, 0, len(ports)) + for _, port := range ports { + if port.Port == nil { + continue + } + values = append(values, *port.Port) + } + return values +}