[kube-proxy] Harden change tracker and proxiers for unmatched IP versions

pkg/proxy/endpoints.go

@@ -29,6 +29,7 @@ import (
 	"k8s.io/client-go/tools/record"
 	api "k8s.io/kubernetes/pkg/apis/core"
 	utilproxy "k8s.io/kubernetes/pkg/proxy/util"
+	utilnet "k8s.io/kubernetes/pkg/util/net"
 )
 
 // EndpointInfoCommon contains common endpoint information.
@@ -206,6 +207,13 @@ func (ect *EndpointChangeTracker) endpointsToEndpointsMap(endpoints *api.Endpoin
 					glog.Warningf("ignoring invalid endpoint port %s with empty host", port.Name)
 					continue
 				}
+				// Filter out the incorrect IP version case.
+				if ect.isIPv6Mode != nil && utilnet.IsIPv6String(addr.IP) != *ect.isIPv6Mode {
+					// Emit event on the corresponding service which had a different
+					// IP version than the endpoint.
+					utilproxy.LogAndEmitIncorrectIPVersionEvent(ect.recorder, "endpoints", addr.IP, endpoints.Name, endpoints.Namespace, "")
+					continue
+				}
 				isLocal := addr.NodeName != nil && *addr.NodeName == ect.hostname
 				epInfoCommon := newEndpointInfoCommon(addr.IP, int(port.Port), isLocal)
 				if ect.customizeEndpointInfo != nil {

pkg/proxy/iptables/proxier.go

@@ -309,6 +309,8 @@ func NewProxier(ipt utiliptables.Interface,
 
 	if len(clusterCIDR) == 0 {
 		glog.Warningf("clusterCIDR not specified, unable to distinguish between internal and external traffic")
+	} else if utilnet.IsIPv6CIDR(clusterCIDR) != ipt.IsIpv6() {
+		return nil, fmt.Errorf("clusterCIDR %s has incorrect IP version: expect isIPv6=%t", clusterCIDR, ipt.IsIpv6())
 	}
 
 	healthChecker := healthcheck.NewServer(hostname, recorder, nil, nil) // use default implementations of deps

pkg/proxy/ipvs/proxier.go

@@ -296,6 +296,8 @@ func NewProxier(ipt utiliptables.Interface,
 
 	if len(clusterCIDR) == 0 {
 		glog.Warningf("clusterCIDR not specified, unable to distinguish between internal and external traffic")
+	} else if utilnet.IsIPv6CIDR(clusterCIDR) != isIPv6 {
+		return nil, fmt.Errorf("clusterCIDR %s has incorrect IP version: expect isIPv6=%t", clusterCIDR, isIPv6)
 	}
 
 	if len(scheduler) == 0 {

pkg/proxy/service.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"net"
 	"reflect"
+	"strings"
 	"sync"
 
 	"github.com/golang/glog"
@@ -31,6 +32,7 @@ import (
 	api "k8s.io/kubernetes/pkg/apis/core"
 	"k8s.io/kubernetes/pkg/apis/core/helper"
 	utilproxy "k8s.io/kubernetes/pkg/proxy/util"
+	utilnet "k8s.io/kubernetes/pkg/util/net"
 )
 
 // ServiceInfoCommon contains common service information.
@@ -92,10 +94,23 @@ func (sct *ServiceChangeTracker) newServiceInfoCommon(port *api.ServicePort, ser
 		OnlyNodeLocalEndpoints: onlyNodeLocalEndpoints,
 	}
 
-	info.ExternalIPs = make([]string, len(service.Spec.ExternalIPs))
-	info.LoadBalancerSourceRanges = make([]string, len(service.Spec.LoadBalancerSourceRanges))
-	copy(info.LoadBalancerSourceRanges, service.Spec.LoadBalancerSourceRanges)
-	copy(info.ExternalIPs, service.Spec.ExternalIPs)
+	if sct.isIPv6Mode == nil {
+		info.ExternalIPs = make([]string, len(service.Spec.ExternalIPs))
+		info.LoadBalancerSourceRanges = make([]string, len(service.Spec.LoadBalancerSourceRanges))
+		copy(info.LoadBalancerSourceRanges, service.Spec.LoadBalancerSourceRanges)
+		copy(info.ExternalIPs, service.Spec.ExternalIPs)
+	} else {
+		// Filter out the incorrect IP version case.
+		var incorrectIPs []string
+		info.ExternalIPs, incorrectIPs = utilnet.FilterIncorrectIPVersion(service.Spec.ExternalIPs, *sct.isIPv6Mode)
+		if len(incorrectIPs) > 0 {
+			utilproxy.LogAndEmitIncorrectIPVersionEvent(sct.recorder, "externalIPs", strings.Join(incorrectIPs, ","), service.Namespace, service.Name, service.UID)
+		}
+		info.LoadBalancerSourceRanges, incorrectIPs = utilnet.FilterIncorrectCIDRVersion(service.Spec.LoadBalancerSourceRanges, *sct.isIPv6Mode)
+		if len(incorrectIPs) > 0 {
+			utilproxy.LogAndEmitIncorrectIPVersionEvent(sct.recorder, "loadBalancerSourceRanges", strings.Join(incorrectIPs, ","), service.Namespace, service.Name, service.UID)
+		}
+	}
 
 	if apiservice.NeedsHealthCheck(service) {
 		p := service.Spec.HealthCheckNodePort
@@ -221,6 +236,14 @@ func (sct *ServiceChangeTracker) serviceToServiceMap(service *api.Service) Servi
 		return nil
 	}
 
+	if len(service.Spec.ClusterIP) != 0 {
+		// Filter out the incorrect IP version case.
+		if sct.isIPv6Mode != nil && utilnet.IsIPv6String(service.Spec.ClusterIP) != *sct.isIPv6Mode {
+			utilproxy.LogAndEmitIncorrectIPVersionEvent(sct.recorder, "clusterIP", service.Spec.ClusterIP, service.Namespace, service.Name, service.UID)
+			return nil
+		}
+	}
+
 	serviceMap := make(ServiceMap)
 	for i := range service.Spec.Ports {
 		servicePort := &service.Spec.Ports[i]