diff --git a/net/bridge.go b/net/bridge.go index c7347ccf..f2772a45 100644 --- a/net/bridge.go +++ b/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 diff --git a/npc/constants.go b/npc/constants.go index 7ab01f7b..2722c57f 100644 --- a/npc/constants.go +++ b/npc/constants.go @@ -8,4 +8,6 @@ const ( IngressChain = "WEAVE-NPC-INGRESS" IpsetNamePrefix = "weave-" + + LocalIpset = IpsetNamePrefix + "local-pods" ) diff --git a/npc/controller.go b/npc/controller.go index c7504be6..550fe31a 100644 --- a/npc/controller.go +++ b/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") diff --git a/npc/ipset/ipset.go b/npc/ipset/ipset.go index d4dd5160..b88e05b4 100644 --- a/npc/ipset/ipset.go +++ b/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 } diff --git a/npc/namespace.go b/npc/namespace.go index 65b00eec..e4d85615 100644 --- a/npc/namespace.go +++ b/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) } diff --git a/npc/selector.go b/npc/selector.go index 1f34fa3b..ba897ad1 100644 --- a/npc/selector.go +++ b/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) } diff --git a/prog/weave-npc/main.go b/prog/weave-npc/main.go index e6818875..c3d3e14e 100644 --- a/prog/weave-npc/main.go +++ b/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()) } diff --git a/test/840_weave_kube_3_test.sh b/test/840_weave_kube_3_test.sh index 4e4fef80..0dc4162f 100755 --- a/test/840_weave_kube_3_test.sh +++ b/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