sdn: handle offset>0 fragments when validating service traffic

By default (set-frag normal) all fragments of a fragmented packet
have a port number of 0.  This is quite unhelpful; to get the right
port number for all fragments requires un-fragmenting the packet
with OVS's contrack capability, but this interacts badly with
iptables' contrack capability.

Instead, use the 'nx-match' mode which keeps the port numbers in
the first fragment of a fragmented packet, and add rules to allow
subsequent fragments (with port=0) to pass through.  Assume that
the destination IP stack will reject offset>0 fragments that
arrive without a corresponding offset=0 first fragment.

We can't just drop the port checking because services can be
manually created with a static service IP address, and perhaps
users rely on creating two distinct services with the same service
IP address, but differentiated via port numbers.

Fixes: https://bugzilla.redhat.com/show_bug.cgi?id=1419692

Author: dcbw

pkg/sdn/plugin/controller.go

@@ -19,6 +19,7 @@ import (
 	utildbus "k8s.io/kubernetes/pkg/util/dbus"
 	kexec "k8s.io/kubernetes/pkg/util/exec"
 	"k8s.io/kubernetes/pkg/util/iptables"
+	"k8s.io/kubernetes/pkg/util/sets"
 	"k8s.io/kubernetes/pkg/util/sysctl"
 	utilwait "k8s.io/kubernetes/pkg/util/wait"
 )
@@ -225,6 +226,10 @@ func (plugin *OsdnNode) SetupSDN() (bool, error) {
 	if err != nil {
 		return false, err
 	}
+	err = plugin.ovs.SetFrags("nx-match")
+	if err != nil {
+		return false, err
+	}
 	_ = plugin.ovs.DeletePort(VXLAN)
 	_, err = plugin.ovs.AddPort(VXLAN, 1, "type=vxlan", `options:remote_ip="flow"`, `options:key="flow"`)
 	if err != nil {
@@ -456,36 +461,56 @@ func (plugin *OsdnNode) DeleteHostSubnetRules(subnet *osapi.HostSubnet) {
 func (plugin *OsdnNode) AddServiceRules(service *kapi.Service, netID uint32) {
 	glog.V(5).Infof("AddServiceRules for %v", service)
 
+	// Use a set to de-duplicate frag rules
+	fragRules := sets.NewString()
+
 	otx := plugin.ovs.NewTransaction()
 	for _, port := range service.Spec.Ports {
-		otx.AddFlow(generateAddServiceRule(netID, service.Spec.ClusterIP, port.Protocol, int(port.Port)))
-		if err := otx.EndTransaction(); err != nil {
-			glog.Errorf("Error adding OVS flows for service %v, netid %d: %v", service, netID, err)
-		}
+		portRule, fragRule := generateAddServiceRules(netID, service.Spec.ClusterIP, port.Protocol, int(port.Port))
+		fragRules.Insert(fragRule)
+		otx.AddFlow(portRule)
+	}
+	for _, rule := range fragRules.List() {
+		otx.AddFlow(rule)
+	}
+	if err := otx.EndTransaction(); err != nil {
+		glog.Errorf("Error adding OVS flows for service %v, netid %d: %v", service, netID, err)
 	}
 }
 
 func (plugin *OsdnNode) DeleteServiceRules(service *kapi.Service) {
 	glog.V(5).Infof("DeleteServiceRules for %v", service)
 
+	// Use a set to de-duplicate frag rules
+	fragRules := sets.NewString()
+
 	otx := plugin.ovs.NewTransaction()
 	for _, port := range service.Spec.Ports {
-		otx.DeleteFlows(generateDeleteServiceRule(service.Spec.ClusterIP, port.Protocol, int(port.Port)))
-		if err := otx.EndTransaction(); err != nil {
-			glog.Errorf("Error deleting OVS flows for service %v: %v", service, err)
-		}
+		portRule, fragRule := generateDeleteServiceRule(service.Spec.ClusterIP, port.Protocol, int(port.Port))
+		fragRules.Insert(fragRule)
+		otx.DeleteFlows(portRule)
+	}
+	for _, rule := range fragRules.List() {
+		otx.DeleteFlows(rule)
+	}
+	if err := otx.EndTransaction(); err != nil {
+		glog.Errorf("Error deleting OVS flows for service %v: %v", service, err)
 	}
 }
 
-func generateBaseServiceRule(IP string, protocol kapi.Protocol, port int) string {
-	return fmt.Sprintf("table=60, %s, nw_dst=%s, tp_dst=%d", strings.ToLower(string(protocol)), IP, port)
+func generateBaseServiceRules(IP string, protocol kapi.Protocol, port int) (string, string) {
+	baseMatch := fmt.Sprintf("table=60, %s, nw_dst=%s", strings.ToLower(string(protocol)), IP)
+	portRule := baseMatch + fmt.Sprintf(", tp_dst=%d", port)
+	fragRule := baseMatch + ", ip_frag=later"
+	return portRule, fragRule
 }
 
-func generateAddServiceRule(netID uint32, IP string, protocol kapi.Protocol, port int) string {
-	baseRule := generateBaseServiceRule(IP, protocol, port)
-	return fmt.Sprintf("%s, priority=100, actions=load:%d->NXM_NX_REG1[], load:2->NXM_NX_REG2[], goto_table:80", baseRule, netID)
+func generateAddServiceRules(netID uint32, IP string, protocol kapi.Protocol, port int) (string, string) {
+	portRule, fragRule := generateBaseServiceRules(IP, protocol, port)
+	actions := fmt.Sprintf(", priority=100, actions=load:%d->NXM_NX_REG1[], load:2->NXM_NX_REG2[], goto_table:80", netID)
+	return portRule + actions, fragRule + actions
 }
 
-func generateDeleteServiceRule(IP string, protocol kapi.Protocol, port int) string {
-	return generateBaseServiceRule(IP, protocol, port)
+func generateDeleteServiceRule(IP string, protocol kapi.Protocol, port int) (string, string) {
+	return generateBaseServiceRules(IP, protocol, port)
 }

pkg/util/ovs/ovs.go

@@ -144,6 +144,11 @@ func (ovsif *Interface) DeletePort(port string) error {
 	return err
 }
 
+func (ovsif *Interface) SetFrags(mode string) error {
+	_, err := ovsif.exec(OVS_OFCTL, "set-frags", ovsif.bridge, mode)
+	return err
+}
+
 type Transaction struct {
 	ovsif *Interface
 	err   error