Merge pull request #2979 from weaveworks/issues/2622-npc-local-node-only

weave-npc to check local addresses only

net/bridge.go

@@ -453,14 +453,14 @@ func configureIPTables(config *BridgeConfig) error {
 		if err = ipt.AppendUnique("filter", "FORWARD", "-i", config.WeaveBridgeName, "-o", config.WeaveBridgeName, "-j", "ACCEPT"); err != nil {
 			return err
 		}
-		// Forward from weave to the rest of the world
-		if err = ipt.AppendUnique("filter", "FORWARD", "-i", config.WeaveBridgeName, "!", "-o", config.WeaveBridgeName, "-j", "ACCEPT"); err != nil {
-			return err
-		}
-		// and allow replies back
-		if err = ipt.AppendUnique("filter", "FORWARD", "-o", config.WeaveBridgeName, "-m", "conntrack", "--ctstate", "RELATED,ESTABLISHED", "-j", "ACCEPT"); err != nil {
-			return err
-		}
+	}
+	// Forward from weave to the rest of the world
+	if err = ipt.AppendUnique("filter", "FORWARD", "-i", config.WeaveBridgeName, "!", "-o", config.WeaveBridgeName, "-j", "ACCEPT"); err != nil {
+		return err
+	}
+	// and allow replies back
+	if err = ipt.AppendUnique("filter", "FORWARD", "-o", config.WeaveBridgeName, "-m", "conntrack", "--ctstate", "RELATED,ESTABLISHED", "-j", "ACCEPT"); err != nil {
+		return err
 	}
 
 	// create a chain for masquerading

npc/constants.go

@@ -8,4 +8,6 @@ const (
 	IngressChain = "WEAVE-NPC-INGRESS"
 
 	IpsetNamePrefix = "weave-"
+
+	LocalIpset = IpsetNamePrefix + "local-pods"
 )

npc/controller.go

@@ -29,6 +29,8 @@ type NetworkPolicyController interface {
 type controller struct {
 	sync.Mutex
 
+	nodeName string // my node name
+
 	ipt *iptables.IPTables
 	ips ipset.Interface
 
@@ -36,11 +38,12 @@ type controller struct {
 	nsSelectors *selectorSet   // selector string -> nsSelector
 }
 
-func New(ipt *iptables.IPTables, ips ipset.Interface) NetworkPolicyController {
+func New(nodeName string, ipt *iptables.IPTables, ips ipset.Interface) NetworkPolicyController {
 	c := &controller{
-		ipt: ipt,
-		ips: ips,
-		nss: make(map[string]*ns)}
+		nodeName: nodeName,
+		ipt:      ipt,
+		ips:      ips,
+		nss:      make(map[string]*ns)}
 
 	c.nsSelectors = newSelectorSet(ips, c.onNewNsSelector)
 
@@ -82,10 +85,21 @@ func (npc *controller) withNS(name string, f func(ns *ns) error) error {
 	return nil
 }
 
+func (npc *controller) checkLocalPod(obj *coreapi.Pod, event string) bool {
+	if obj.Spec.NodeName != npc.nodeName {
+		common.Log.Debugf("%s ignored for pod %s/%s on node %s", event, obj.ObjectMeta.Namespace, obj.ObjectMeta.Name, obj.Spec.NodeName)
+		return false
+	}
+	return true
+}
+
 func (npc *controller) AddPod(obj *coreapi.Pod) error {
 	npc.Lock()
 	defer npc.Unlock()
 
+	if !npc.checkLocalPod(obj, "AddPod") {
+		return nil
+	}
 	common.Log.Debugf("EVENT AddPod %s", js(obj))
 	return npc.withNS(obj.ObjectMeta.Namespace, func(ns *ns) error {
 		return errors.Wrap(ns.addPod(obj), "add pod")
@@ -96,6 +110,10 @@ func (npc *controller) UpdatePod(oldObj, newObj *coreapi.Pod) error {
 	npc.Lock()
 	defer npc.Unlock()
 
+	// Pods don't move across nodes so we only check the new version
+	if !npc.checkLocalPod(newObj, "UpdatePod") {
+		return nil
+	}
 	common.Log.Debugf("EVENT UpdatePod %s %s", js(oldObj), js(newObj))
 	return npc.withNS(oldObj.ObjectMeta.Namespace, func(ns *ns) error {
 		return errors.Wrap(ns.updatePod(oldObj, newObj), "update pod")
@@ -106,6 +124,9 @@ func (npc *controller) DeletePod(obj *coreapi.Pod) error {
 	npc.Lock()
 	defer npc.Unlock()
 
+	if !npc.checkLocalPod(obj, "DeletePod") {
+		return nil
+	}
 	common.Log.Debugf("EVENT DeletePod %s", js(obj))
 	return npc.withNS(obj.ObjectMeta.Namespace, func(ns *ns) error {
 		return errors.Wrap(ns.deletePod(obj), "delete pod")

npc/ipset/ipset.go

@@ -1,6 +1,7 @@
 package ipset
 
 import (
+	"log"
 	"os/exec"
 	"strings"
 
@@ -31,10 +32,11 @@ type Interface interface {
 
 type ipset struct {
 	refCount
+	*log.Logger
 }
 
-func New() Interface {
-	return &ipset{refCount: newRefCount()}
+func New(logger *log.Logger) Interface {
+	return &ipset{refCount: newRefCount(), Logger: logger}
 }
 
 func (i *ipset) Create(ipsetName Name, ipsetType Type) error {
@@ -42,6 +44,7 @@ func (i *ipset) Create(ipsetName Name, ipsetType Type) error {
 }
 
 func (i *ipset) AddEntry(ipsetName Name, entry string) error {
+	i.Logger.Printf("adding entry %s to %s", entry, ipsetName)
 	if i.inc(ipsetName, entry) > 1 { // already in the set
 		return nil
 	}
@@ -49,6 +52,7 @@ func (i *ipset) AddEntry(ipsetName Name, entry string) error {
 }
 
 func (i *ipset) DelEntry(ipsetName Name, entry string) error {
+	i.Logger.Printf("deleting entry %s from %s", entry, ipsetName)
 	if i.dec(ipsetName, entry) > 0 { // still needed
 		return nil
 	}

npc/namespace.go

@@ -88,6 +88,7 @@ func (ns *ns) addPod(obj *coreapi.Pod) error {
 		return nil
 	}
 
+	ns.ips.AddEntry(LocalIpset, obj.Status.PodIP)
 	return ns.podSelectors.addToMatching(obj.ObjectMeta.Labels, obj.Status.PodIP)
 }
 
@@ -100,10 +101,12 @@ func (ns *ns) updatePod(oldObj, newObj *coreapi.Pod) error {
 	}
 
 	if hasIP(oldObj) && !hasIP(newObj) {
+		ns.ips.DelEntry(LocalIpset, oldObj.Status.PodIP)
 		return ns.podSelectors.delFromMatching(oldObj.ObjectMeta.Labels, oldObj.Status.PodIP)
 	}
 
 	if !hasIP(oldObj) && hasIP(newObj) {
+		ns.ips.AddEntry(LocalIpset, newObj.Status.PodIP)
 		return ns.podSelectors.addToMatching(newObj.ObjectMeta.Labels, newObj.Status.PodIP)
 	}
 
@@ -139,6 +142,7 @@ func (ns *ns) deletePod(obj *coreapi.Pod) error {
 		return nil
 	}
 
+	ns.ips.DelEntry(LocalIpset, obj.Status.PodIP)
 	return ns.podSelectors.delFromMatching(obj.ObjectMeta.Labels, obj.Status.PodIP)
 }
 

npc/selector.go

@@ -43,12 +43,10 @@ func (s *selector) matches(labelMap map[string]string) bool {
 }
 
 func (s *selector) addEntry(entry string) error {
-	common.Log.Infof("adding entry %s to %s", entry, s.spec.ipsetName)
 	return s.ips.AddEntry(s.spec.ipsetName, entry)
 }
 
 func (s *selector) delEntry(entry string) error {
-	common.Log.Infof("deleting entry %s from %s", entry, s.spec.ipsetName)
 	return s.ips.DelEntry(s.spec.ipsetName, entry)
 }
 

prog/weave-npc/main.go

@@ -28,6 +28,7 @@ var (
 	metricsAddr string
 	logLevel    string
 	allowMcast  bool
+	nodeName    string
 )
 
 func handleError(err error) { common.CheckFatal(err) }
@@ -53,29 +54,6 @@ func resetIPTables(ipt *iptables.IPTables) error {
 		return err
 	}
 
-	// Configure main chain static rules
-	if err := ipt.Append(npc.TableFilter, npc.MainChain,
-		"-m", "state", "--state", "RELATED,ESTABLISHED", "-j", "ACCEPT"); err != nil {
-		return err
-	}
-
-	if allowMcast {
-		if err := ipt.Append(npc.TableFilter, npc.MainChain,
-			"-d", "224.0.0.0/4", "-j", "ACCEPT"); err != nil {
-			return err
-		}
-	}
-
-	if err := ipt.Append(npc.TableFilter, npc.MainChain,
-		"-m", "state", "--state", "NEW", "-j", string(npc.DefaultChain)); err != nil {
-		return err
-	}
-
-	if err := ipt.Append(npc.TableFilter, npc.MainChain,
-		"-m", "state", "--state", "NEW", "-j", string(npc.IngressChain)); err != nil {
-		return err
-	}
-
 	return nil
 }
 
@@ -110,9 +88,53 @@ func resetIPSets(ips ipset.Interface) error {
 	return nil
 }
 
+func createBaseRules(ipt *iptables.IPTables, ips ipset.Interface) error {
+	// Configure main chain static rules
+	if err := ipt.Append(npc.TableFilter, npc.MainChain,
+		"-m", "state", "--state", "RELATED,ESTABLISHED", "-j", "ACCEPT"); err != nil {
+		return err
+	}
+
+	if allowMcast {
+		if err := ipt.Append(npc.TableFilter, npc.MainChain,
+			"-d", "224.0.0.0/4", "-j", "ACCEPT"); err != nil {
+			return err
+		}
+	}
+
+	if err := ipt.Append(npc.TableFilter, npc.MainChain,
+		"-m", "state", "--state", "NEW", "-j", string(npc.DefaultChain)); err != nil {
+		return err
+	}
+
+	if err := ipt.Append(npc.TableFilter, npc.MainChain,
+		"-m", "state", "--state", "NEW", "-j", string(npc.IngressChain)); err != nil {
+		return err
+	}
+
+	// If the destination address is not any of the local pods, let it through
+	if err := ips.Create(npc.LocalIpset, ipset.HashIP); err != nil {
+		return err
+	}
+	if err := ipt.Append(npc.TableFilter, npc.MainChain,
+		"-m", "set", "--match-set", npc.LocalIpset, "src",
+		"-m", "set", "!", "--match-set", npc.LocalIpset, "dst", "-j", "ACCEPT"); err != nil {
+		return err
+	}
+
+	return nil
+}
+
 func root(cmd *cobra.Command, args []string) {
 	common.SetLogLevel(logLevel)
-	common.Log.Infof("Starting Weaveworks NPC %s", version)
+	if nodeName == "" {
+		// HOSTNAME is set by Kubernetes for pods in the host network namespace
+		nodeName = os.Getenv("HOSTNAME")
+	}
+	if nodeName == "" {
+		common.Log.Fatalf("Must set node name via --node-name or $HOSTNAME")
+	}
+	common.Log.Infof("Starting Weaveworks NPC %s; node name %q", version, nodeName)
 
 	if err := metrics.Start(metricsAddr); err != nil {
 		common.Log.Fatalf("Failed to start metrics: %v", err)
@@ -131,12 +153,13 @@ func root(cmd *cobra.Command, args []string) {
 	ipt, err := iptables.New()
 	handleError(err)
 
-	ips := ipset.New()
+	ips := ipset.New(common.LogLogger())
 
 	handleError(resetIPTables(ipt))
 	handleError(resetIPSets(ips))
+	handleError(createBaseRules(ipt, ips))
 
-	npc := npc.New(ipt, ips)
+	npc := npc.New(nodeName, ipt, ips)
 
 	nsController := makeController(client.Core().RESTClient(), "namespaces", &coreapi.Namespace{},
 		cache.ResourceEventHandlerFuncs{
@@ -216,6 +239,7 @@ func main() {
 	rootCmd.PersistentFlags().StringVar(&metricsAddr, "metrics-addr", ":6781", "metrics server bind address")
 	rootCmd.PersistentFlags().StringVar(&logLevel, "log-level", "debug", "logging level (debug, info, warning, error)")
 	rootCmd.PersistentFlags().BoolVar(&allowMcast, "allow-mcast", true, "allow all multicast traffic")
+	rootCmd.PersistentFlags().StringVar(&nodeName, "node-name", "", "only generate rules that apply to this node")
 
 	handleError(rootCmd.Execute())
 }

test/840_weave_kube_3_test.sh

@@ -101,6 +101,10 @@ check_all_pods_communicate() {
 
 assert_raises 'wait_for_x check_all_pods_communicate pods'
 
+# Check that a pod can contact the outside world
+podName=$($SSH $HOST1 "$KUBECTL get pods -l run=nettest -o go-template='{{(index .items 0).metadata.name}}'")
+assert_raises "$SSH $HOST1 $KUBECTL exec $podName -- $PING 8.8.8.8"
+
 tear_down_kubeadm
 
 # Destroy our test ipset, and implicitly check it is still there