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 +}