diff --git a/.gitignore b/.gitignore index 4c58507e..aef563bc 100644 --- a/.gitignore +++ b/.gitignore @@ -46,6 +46,7 @@ prog/weaveutil/weaveutil prog/*/Dockerfile.weaveworks prog/kube-peers/kube-peers prog/weave-kube/kube-peers +prog/weave-npc/weave-npc prog/plugin/plugin testing/cover/cover testing/runner/runner diff --git a/.gitmodules b/.gitmodules index 75655c9f..764d4bff 100644 --- a/.gitmodules +++ b/.gitmodules @@ -106,3 +106,15 @@ [submodule "vendor/k8s.io/client-go"] path = vendor/k8s.io/client-go url = https://github.com/kubernetes/client-go +[submodule "vendor/github.com/pkg/errors"] + path = vendor/github.com/pkg/errors + url = https://github.com/pkg/errors +[submodule "vendor/github.com/spf13/cobra"] + path = vendor/github.com/spf13/cobra + url = https://github.com/spf13/cobra +[submodule "vendor/github.com/spf13/plag"] + path = vendor/github.com/spf13/plag + url = https://github.com/spf13/pflag +[submodule "vendor/github.com/spf13/pflag"] + path = vendor/github.com/spf13/pflag + url = https://github.com/spf13/pflag diff --git a/Makefile b/Makefile index ad82d59a..8aeea2af 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -PUBLISH=publish_weave publish_weaveexec publish_plugin publish_weave-kube +PUBLISH=publish_weave publish_weaveexec publish_plugin publish_weave-kube publish_weave-npc .DEFAULT: all .PHONY: all exes testrunner update tests lint publish $(PUBLISH) clean clean-bin prerequisites build run-smoketests @@ -17,6 +17,7 @@ WEAVER_EXE=prog/weaver/weaver WEAVEPROXY_EXE=prog/weaveproxy/weaveproxy SIGPROXY_EXE=prog/sigproxy/sigproxy KUBEPEERS_EXE=prog/kube-peers/kube-peers +WEAVENPC_EXE=prog/weave-npc/weave-npc WEAVEWAIT_EXE=prog/weavewait/weavewait WEAVEWAIT_NOOP_EXE=prog/weavewait/weavewait_noop WEAVEWAIT_NOMCAST_EXE=prog/weavewait/weavewait_nomcast @@ -25,25 +26,27 @@ PLUGIN_EXE=prog/plugin/plugin RUNNER_EXE=tools/runner/runner TEST_TLS_EXE=test/tls/tls -EXES=$(WEAVER_EXE) $(SIGPROXY_EXE) $(KUBEPEERS_EXE) $(WEAVEPROXY_EXE) $(WEAVEWAIT_EXE) $(WEAVEWAIT_NOOP_EXE) $(WEAVEWAIT_NOMCAST_EXE) $(WEAVEUTIL_EXE) $(PLUGIN_EXE) $(TEST_TLS_EXE) +EXES=$(WEAVER_EXE) $(SIGPROXY_EXE) $(KUBEPEERS_EXE) $(WEAVENPC_EXE) $(WEAVEPROXY_EXE) $(WEAVEWAIT_EXE) $(WEAVEWAIT_NOOP_EXE) $(WEAVEWAIT_NOMCAST_EXE) $(WEAVEUTIL_EXE) $(PLUGIN_EXE) $(TEST_TLS_EXE) BUILD_UPTODATE=.build.uptodate WEAVER_UPTODATE=.weaver.uptodate WEAVEEXEC_UPTODATE=.weaveexec.uptodate PLUGIN_UPTODATE=.plugin.uptodate WEAVEKUBE_UPTODATE=.weavekube.uptodate +WEAVENPC_UPTODATE=.weavenpc.uptodate WEAVEDB_UPTODATE=.weavedb.uptodate -IMAGES_UPTODATE=$(WEAVER_UPTODATE) $(WEAVEEXEC_UPTODATE) $(PLUGIN_UPTODATE) $(WEAVEKUBE_UPTODATE) $(WEAVEDB_UPTODATE) +IMAGES_UPTODATE=$(WEAVER_UPTODATE) $(WEAVEEXEC_UPTODATE) $(PLUGIN_UPTODATE) $(WEAVEKUBE_UPTODATE) $(WEAVENPC_UPTODATE) $(WEAVEDB_UPTODATE) WEAVER_IMAGE=$(DOCKERHUB_USER)/weave WEAVEEXEC_IMAGE=$(DOCKERHUB_USER)/weaveexec PLUGIN_IMAGE=$(DOCKERHUB_USER)/plugin WEAVEKUBE_IMAGE=$(DOCKERHUB_USER)/weave-kube +WEAVENPC_IMAGE=$(DOCKERHUB_USER)/weave-npc BUILD_IMAGE=$(DOCKERHUB_USER)/weavebuild WEAVEDB_IMAGE=$(DOCKERHUB_USER)/weavedb -IMAGES=$(WEAVER_IMAGE) $(WEAVEEXEC_IMAGE) $(PLUGIN_IMAGE) $(WEAVEKUBE_IMAGE) $(WEAVEDB_IMAGE) +IMAGES=$(WEAVER_IMAGE) $(WEAVEEXEC_IMAGE) $(PLUGIN_IMAGE) $(WEAVEKUBE_IMAGE) $(WEAVENPC_IMAGE) $(WEAVEDB_IMAGE) WEAVE_EXPORT=weave.tar.gz @@ -71,6 +74,7 @@ $(WEAVEPROXY_EXE): proxy/*.go prog/weaveproxy/*.go $(WEAVEUTIL_EXE): prog/weaveutil/*.go net/*.go $(SIGPROXY_EXE): prog/sigproxy/*.go $(KUBEPEERS_EXE): prog/kube-peers/*.go +$(WEAVENPC_EXE): prog/weave-npc/*.go npc/*.go npc/*/*.go $(PLUGIN_EXE): prog/plugin/*.go plugin/*/*.go api/*.go common/*.go common/docker/*.go net/*.go $(TEST_TLS_EXE): test/tls/*.go $(WEAVEWAIT_NOOP_EXE): prog/weavewait/*.go @@ -103,7 +107,7 @@ else endif $(NETGO_CHECK) -$(WEAVEUTIL_EXE) $(KUBEPEERS_EXE): +$(WEAVEUTIL_EXE) $(KUBEPEERS_EXE) $(WEAVENPC_EXE): go build $(BUILD_FLAGS) -o $@ ./$(@D) $(NETGO_CHECK) @@ -160,6 +164,10 @@ $(WEAVEKUBE_UPTODATE): prog/weave-kube/Dockerfile.$(DOCKERHUB_USER) prog/weave-k $(SUDO) docker build -f prog/weave-kube/Dockerfile.$(DOCKERHUB_USER) -t $(WEAVEKUBE_IMAGE) prog/weave-kube touch $@ +$(WEAVENPC_UPTODATE): prog/weave-npc/Dockerfile $(WEAVENPC_EXE) prog/weave-npc/ulogd.conf + $(SUDO) docker build -f prog/weave-npc/Dockerfile -t $(WEAVENPC_IMAGE) prog/weave-npc + touch $@ + $(WEAVEDB_UPTODATE): prog/weavedb/Dockerfile $(SUDO) docker build -t $(WEAVEDB_IMAGE) prog/weavedb touch $@ diff --git a/docs/weavenpc-design.md b/docs/weavenpc-design.md new file mode 100644 index 00000000..5b1693bf --- /dev/null +++ b/docs/weavenpc-design.md @@ -0,0 +1,100 @@ +# Overview + +# ipsets + +The policy controller maintains a number of ipsets which are +subsequently referred to by the iptables rules used to effect network +policy specifications. These ipsets are created, modified and +destroyed automatically in response to Pod, Namespace and +NetworkPolicy object updates from the k8s API server: + +* A `hash:ip` set per namespace, containing the IP addresses of all + pods in that namespace +* A `list:set` per distinct (across all network policies in all + namespaces) namespace selector mentioned in a network policy, + containing the names of any of the above hash:ip sets whose + corresponding namespace labels match the selector +* A `hash:ip` set for each distinct (within the scope of the + containing network policy's namespace) pod selector mentioned in a + network policy, containing the IP addresses of all pods in the + namespace whose labels match that selector + +ipset names are generated deterministically from a string +representation of the corresponding label selector. Because ipset +names are limited to 31 characters in length, this is done by taking a +SHA hash of the selector string and then printing that out as a base +85 string with a "weave-" prefix e.g.: + + weave-k?Z;25^M}|1s7P3|H9i;*;MhG + +Because pod selectors are scoped to a namespace, we need to make sure +that if the same selector definition is used in different namespaces +that we maintain distinct ipsets. Consequently, for such selectors the +namespace name is prepended to the label selector string before +hashing to avoid clashes. + +# iptables chains + +The policy controller maintains two iptables chains in response to +changes to pods, namespaces and network policies. One chain contains +the ingress rules that implement the network policy specifications, +and the other is used to bypass the ingress rules for namespaces which +have an ingress isolation policy of `DefaultAllow`. + +## Dynamically maintained `WEAVE-NPC-DEFAULT` chain + +The policy controller maintains a rule in this chain for every +namespace whose ingress isolation policy is `DefaultAllow`. The +purpose of this rule is simply to ACCEPT any traffic destined for such +namespaces before it reaches the ingress chain. + +``` +iptables -A WEAVE-NPC-DEFAULT -m set --match-set $NSIPSET dst -j ACCEPT +``` + +## Dynamically maintained `WEAVE-NPC-INGRESS` chain + +For each namespace network policy ingress rule peer/port combination: + +``` +iptables -A WEAVE-NPC-INGRESS -p $PROTO [-m set --match-set $SRCSET] -m set --match-set $DSTSET --dport $DPORT -j ACCEPT +``` + +## Static `WEAVE-NPC` chain + +Static configuration: + +``` +iptables -A WEAVE-NPC -m state --state RELATED,ESTABLISHED -j ACCEPT +iptables -A WEAVE-NPC -m state --state NEW -j WEAVE-NPC-DEFAULT +iptables -A WEAVE-NPC -m state --state NEW -j WEAVE-NPC-INGRESS +``` + +# Steering traffic into the policy engine + +To direct traffic into the policy engine: + +iptables -A FORWARD -o weave -m physdev ! --physdev-out vethwe-bridge -j WEAVE-NPC +iptables -A FORWARD -o weave -m physdev ! --physdev-out vethwe-bridge -j DROP + +Note this only affects traffic which egresses the bridge on a physical +port which is not the Weave Net router - in other words, it is +destined for an application container veth. The following traffic is +affected: + +* Traffic bridged between local application containers +* Traffic bridged from the router to a local application container +* Traffic originating from the internet destined for nodeports - this + is routed via the FORWARD chain to a container pod IP after DNAT + +The following traffic is NOT affected: + +* Traffic bridged from a local application container to the router +* Traffic originating from processes in the host network namespace + (e.g. kubelet health checks) +* Traffic routed from an application container to the internet + +See these resources for helpful context: + +* http://ebtables.netfilter.org/br_fw_ia/br_fw_ia.html +* https://commons.wikimedia.org/wiki/File:Netfilter-packet-flow.svg diff --git a/npc/analyser.go b/npc/analyser.go new file mode 100644 index 00000000..312d529c --- /dev/null +++ b/npc/analyser.go @@ -0,0 +1,114 @@ +package npc + +import ( + "fmt" + + "k8s.io/client-go/pkg/api" + extnapi "k8s.io/client-go/pkg/apis/extensions/v1beta1" + "k8s.io/client-go/pkg/util/intstr" + + "github.com/weaveworks/weave/npc/ipset" +) + +func (ns *ns) analysePolicy(policy *extnapi.NetworkPolicy) ( + rules map[string]*ruleSpec, + nsSelectors, podSelectors map[string]*selectorSpec, + err error) { + + nsSelectors = make(map[string]*selectorSpec) + podSelectors = make(map[string]*selectorSpec) + rules = make(map[string]*ruleSpec) + + dstSelector, err := newSelectorSpec(&policy.Spec.PodSelector, ns.name, ipset.HashIP) + if err != nil { + return nil, nil, nil, err + } + podSelectors[dstSelector.key] = dstSelector + + for _, ingressRule := range policy.Spec.Ingress { + if ingressRule.Ports != nil && len(ingressRule.Ports) == 0 { + // Ports is empty, this rule matches no ports (no traffic matches). + continue + } + + if ingressRule.From != nil && len(ingressRule.From) == 0 { + // From is empty, this rule matches no sources (no traffic matches). + continue + } + + if ingressRule.From == nil { + // From is not provided, this rule matches all sources (traffic not restricted by source). + if ingressRule.Ports == nil { + // Ports is not provided, this rule matches all ports (traffic not restricted by port). + rule := newRuleSpec(nil, nil, dstSelector, nil) + rules[rule.key] = rule + } else { + // Ports is present and contains at least one item, then this rule allows traffic + // only if the traffic matches at least one port in the ports list. + withNormalisedProtoAndPort(ingressRule.Ports, func(proto, port string) { + rule := newRuleSpec(&proto, nil, dstSelector, &port) + rules[rule.key] = rule + }) + } + } else { + // From is present and contains at least on item, this rule allows traffic only if the + // traffic matches at least one item in the from list. + for _, peer := range ingressRule.From { + var srcSelector *selectorSpec + if peer.PodSelector != nil { + srcSelector, err = newSelectorSpec(peer.PodSelector, ns.name, ipset.HashIP) + if err != nil { + return nil, nil, nil, err + } + podSelectors[srcSelector.key] = srcSelector + } + if peer.NamespaceSelector != nil { + srcSelector, err = newSelectorSpec(peer.NamespaceSelector, "", ipset.ListSet) + if err != nil { + return nil, nil, nil, err + } + nsSelectors[srcSelector.key] = srcSelector + } + + if ingressRule.Ports == nil { + // Ports is not provided, this rule matches all ports (traffic not restricted by port). + rule := newRuleSpec(nil, srcSelector, dstSelector, nil) + rules[rule.key] = rule + } else { + // Ports is present and contains at least one item, then this rule allows traffic + // only if the traffic matches at least one port in the ports list. + withNormalisedProtoAndPort(ingressRule.Ports, func(proto, port string) { + rule := newRuleSpec(&proto, srcSelector, dstSelector, &port) + rules[rule.key] = rule + }) + } + } + } + } + + return rules, nsSelectors, podSelectors, nil +} + +func withNormalisedProtoAndPort(npps []extnapi.NetworkPolicyPort, f func(proto, port string)) { + for _, npp := range npps { + // If no proto is specified, default to TCP + proto := string(api.ProtocolTCP) + if npp.Protocol != nil { + proto = string(*npp.Protocol) + } + + // If no port is specified, match any port. Let iptables executable handle + // service name resolution + port := "0:65535" + if npp.Port != nil { + switch npp.Port.Type { + case intstr.Int: + port = fmt.Sprintf("%d", npp.Port.IntVal) + case intstr.String: + port = npp.Port.StrVal + } + } + + f(proto, port) + } +} diff --git a/npc/betatypes.go b/npc/betatypes.go new file mode 100644 index 00000000..8b8f4441 --- /dev/null +++ b/npc/betatypes.go @@ -0,0 +1,40 @@ +package npc + +// These types are defined in https://github.com/kubernetes/kubernetes/blob/master/docs/proposals/network-policy.md +// but not in the k8s API yet. Copies are included here for decoding the `net.beta.kubernetes.io/network-policy` +// annotation specified in the above document. + +type IngressIsolationPolicy string + +const ( + // Deny all ingress traffic to pods in this namespace. Ingress means + // any incoming traffic to pods, whether that be from other pods within this namespace + // or any source outside of this namespace. + DefaultDeny IngressIsolationPolicy = "DefaultDeny" +) + +// Standard NamespaceSpec object, modified to include a new +// NamespaceNetworkPolicy field. +type NamespaceSpec struct { + // This is a pointer so that it can be left undefined. + NetworkPolicy *NamespaceNetworkPolicy `json:"networkPolicy,omitempty"` +} + +type NamespaceNetworkPolicy struct { + // Ingress configuration for this namespace. This config is + // applied to all pods within this namespace. For now, only + // ingress is supported. This field is optional - if not + // defined, then the cluster default for ingress is applied. + Ingress *NamespaceIngressPolicy `json:"ingress,omitempty"` +} + +// Configuration for ingress to pods within this namespace. +// For now, this only supports specifying an isolation policy. +type NamespaceIngressPolicy struct { + // The isolation policy to apply to pods in this namespace. + // Currently this field only supports "DefaultDeny", but could + // be extended to support other policies in the future. When set to DefaultDeny, + // pods in this namespace are denied ingress traffic by default. When not defined, + // the cluster default ingress isolation policy is applied (currently allow all). + Isolation *IngressIsolationPolicy `json:"isolation,omitempty"` +} diff --git a/npc/constants.go b/npc/constants.go new file mode 100644 index 00000000..64565484 --- /dev/null +++ b/npc/constants.go @@ -0,0 +1,9 @@ +package npc + +const ( + TableFilter = "filter" + + MainChain = "WEAVE-NPC" + DefaultChain = "WEAVE-NPC-DEFAULT" + IngressChain = "WEAVE-NPC-INGRESS" +) diff --git a/npc/controller.go b/npc/controller.go new file mode 100644 index 00000000..c3153c3c --- /dev/null +++ b/npc/controller.go @@ -0,0 +1,173 @@ +package npc + +import ( + "sync" + + log "github.com/Sirupsen/logrus" + "github.com/coreos/go-iptables/iptables" + "github.com/pkg/errors" + coreapi "k8s.io/client-go/pkg/api/v1" + extnapi "k8s.io/client-go/pkg/apis/extensions/v1beta1" + + "github.com/weaveworks/weave/npc/ipset" +) + +type NetworkPolicyController interface { + AddNamespace(ns *coreapi.Namespace) error + UpdateNamespace(oldObj, newObj *coreapi.Namespace) error + DeleteNamespace(ns *coreapi.Namespace) error + + AddPod(obj *coreapi.Pod) error + UpdatePod(oldObj, newObj *coreapi.Pod) error + DeletePod(obj *coreapi.Pod) error + + AddNetworkPolicy(obj *extnapi.NetworkPolicy) error + UpdateNetworkPolicy(oldObj, newObj *extnapi.NetworkPolicy) error + DeleteNetworkPolicy(obj *extnapi.NetworkPolicy) error +} + +type controller struct { + sync.Mutex + + ipt *iptables.IPTables + ips ipset.Interface + + nss map[string]*ns // ns name -> ns struct + nsSelectors *selectorSet // selector string -> nsSelector +} + +func New(ipt *iptables.IPTables, ips ipset.Interface) NetworkPolicyController { + c := &controller{ + ipt: ipt, + ips: ips, + nss: make(map[string]*ns)} + + c.nsSelectors = newSelectorSet(ips, c.onNewNsSelector) + + return c +} + +func (npc *controller) onNewNsSelector(selector *selector) error { + for _, ns := range npc.nss { + if ns.namespace != nil { + if selector.matches(ns.namespace.ObjectMeta.Labels) { + if err := ns.ips.AddEntry(selector.spec.ipsetName, string(ns.allPods.ipsetName)); err != nil { + return err + } + } + } + } + return nil +} + +func (npc *controller) withNS(name string, f func(ns *ns) error) error { + ns, found := npc.nss[name] + if !found { + newNs, err := newNS(name, npc.ipt, npc.ips, npc.nsSelectors) + if err != nil { + return err + } + npc.nss[name] = newNs + ns = newNs + } + if err := f(ns); err != nil { + return err + } + if ns.empty() { + if err := ns.destroy(); err != nil { + return err + } + delete(npc.nss, name) + } + return nil +} + +func (npc *controller) AddPod(obj *coreapi.Pod) error { + npc.Lock() + defer npc.Unlock() + + log.Infof("EVENT AddPod %#v", obj) + return npc.withNS(obj.ObjectMeta.Namespace, func(ns *ns) error { + return errors.Wrap(ns.addPod(obj), "add pod") + }) +} + +func (npc *controller) UpdatePod(oldObj, newObj *coreapi.Pod) error { + npc.Lock() + defer npc.Unlock() + + log.Infof("EVENT UpdatePod %#v %#v", oldObj, newObj) + return npc.withNS(oldObj.ObjectMeta.Namespace, func(ns *ns) error { + return errors.Wrap(ns.updatePod(oldObj, newObj), "update pod") + }) +} + +func (npc *controller) DeletePod(obj *coreapi.Pod) error { + npc.Lock() + defer npc.Unlock() + + log.Infof("EVENT DeletePod %#v", obj) + return npc.withNS(obj.ObjectMeta.Namespace, func(ns *ns) error { + return errors.Wrap(ns.deletePod(obj), "delete pod") + }) +} + +func (npc *controller) AddNetworkPolicy(obj *extnapi.NetworkPolicy) error { + npc.Lock() + defer npc.Unlock() + + log.Infof("EVENT AddNetworkPolicy %#v", obj) + return npc.withNS(obj.ObjectMeta.Namespace, func(ns *ns) error { + return errors.Wrap(ns.addNetworkPolicy(obj), "add network policy") + }) +} + +func (npc *controller) UpdateNetworkPolicy(oldObj, newObj *extnapi.NetworkPolicy) error { + npc.Lock() + defer npc.Unlock() + + log.Infof("EVENT UpdateNetworkPolicy %#v %#v", oldObj, newObj) + return npc.withNS(oldObj.ObjectMeta.Namespace, func(ns *ns) error { + return errors.Wrap(ns.updateNetworkPolicy(oldObj, newObj), "update network policy") + }) +} + +func (npc *controller) DeleteNetworkPolicy(obj *extnapi.NetworkPolicy) error { + npc.Lock() + defer npc.Unlock() + + log.Infof("EVENT DeleteNetworkPolicy %#v", obj) + return npc.withNS(obj.ObjectMeta.Namespace, func(ns *ns) error { + return errors.Wrap(ns.deleteNetworkPolicy(obj), "delete network policy") + }) +} + +func (npc *controller) AddNamespace(obj *coreapi.Namespace) error { + npc.Lock() + defer npc.Unlock() + + log.Infof("EVENT AddNamespace %#v", obj) + return npc.withNS(obj.ObjectMeta.Name, func(ns *ns) error { + return errors.Wrap(ns.addNamespace(obj), "add namespace") + }) +} + +func (npc *controller) UpdateNamespace(oldObj, newObj *coreapi.Namespace) error { + npc.Lock() + defer npc.Unlock() + + log.Infof("EVENT UpdateNamespace%#v %#v", oldObj, newObj) + return npc.withNS(oldObj.ObjectMeta.Name, func(ns *ns) error { + return errors.Wrap(ns.updateNamespace(oldObj, newObj), "update namespace") + }) +} + +func (npc *controller) DeleteNamespace(obj *coreapi.Namespace) error { + npc.Lock() + defer npc.Unlock() + + log.Infof("EVENT DeleteNamespace %#v", obj) + return npc.withNS(obj.ObjectMeta.Name, func(ns *ns) error { + return errors.Wrap(ns.deleteNamespace(obj), "delete namespace") + }) +} diff --git a/npc/ipset/ipset.go b/npc/ipset/ipset.go new file mode 100644 index 00000000..5fbdb3db --- /dev/null +++ b/npc/ipset/ipset.go @@ -0,0 +1,69 @@ +package ipset + +import ( + "os/exec" + + "github.com/pkg/errors" +) + +type Name string + +type Type string + +const ( + ListSet = Type("list:set") + HashIP = Type("hash:ip") +) + +type Interface interface { + Create(ipsetName Name, ipsetType Type) error + AddEntry(ipsetName Name, entry string) error + DelEntry(ipsetName Name, entry string) error + Flush(ipsetName Name) error + Destroy(ipsetName Name) error + + FlushAll() error + DestroyAll() error +} + +type ipset struct { +} + +func New() Interface { + return &ipset{} +} + +func (i *ipset) Create(ipsetName Name, ipsetType Type) error { + return doExec("create", string(ipsetName), string(ipsetType)) +} + +func (i *ipset) AddEntry(ipsetName Name, entry string) error { + return doExec("add", string(ipsetName), entry) +} + +func (i *ipset) DelEntry(ipsetName Name, entry string) error { + return doExec("del", string(ipsetName), entry) +} + +func (i *ipset) Flush(ipsetName Name) error { + return doExec("flush", string(ipsetName)) +} + +func (i *ipset) FlushAll() error { + return doExec("flush") +} + +func (i *ipset) Destroy(ipsetName Name) error { + return doExec("destroy", string(ipsetName)) +} + +func (i *ipset) DestroyAll() error { + return doExec("destroy") +} + +func doExec(args ...string) error { + if output, err := exec.Command("ipset", args...).CombinedOutput(); err != nil { + return errors.Wrapf(err, "ipset %v failed: %s", args, output) + } + return nil +} diff --git a/npc/metrics/metrics.go b/npc/metrics/metrics.go new file mode 100644 index 00000000..9ee8639a --- /dev/null +++ b/npc/metrics/metrics.go @@ -0,0 +1,78 @@ +package metrics + +import ( + log "github.com/Sirupsen/logrus" + "net/http" + "os" + "strconv" + + "github.com/google/gopacket" + "github.com/google/gopacket/layers" + "github.com/google/gopacket/pcapgo" + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promhttp" +) + +var ( + blockedConnections = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Name: "weavenpc_blocked_connections_total", + Help: "Connection attempts blocked by policy controller.", + }, + []string{"protocol", "dport"}, + ) +) + +func gatherMetrics() { + pipe, err := os.Open("/var/log/ulogd.pcap") + if err != nil { + log.Fatalf("Failed to open pcap: %v", err) + } + + reader, err := pcapgo.NewReader(pipe) + if err != nil { + log.Fatalf("Failed to read pcap header: %v", err) + } + + for { + data, _, err := reader.ReadPacketData() + if err != nil { + log.Fatalf("Failed to read pcap packet: %v", err) + } + + packet := gopacket.NewPacket(data, layers.LayerTypeIPv4, gopacket.Default) + + if tcpLayer := packet.Layer(layers.LayerTypeTCP); tcpLayer != nil { + tcp, _ := tcpLayer.(*layers.TCP) + if tcp.SYN && !tcp.ACK { // Only plain SYN constitutes a NEW TCP connection + blockedConnections.With(prometheus.Labels{"protocol": "tcp", "dport": strconv.Itoa(int(tcp.DstPort))}).Inc() + continue + } + } + + if udpLayer := packet.Layer(layers.LayerTypeUDP); udpLayer != nil { + udp, _ := udpLayer.(*layers.UDP) + blockedConnections.With(prometheus.Labels{"protocol": "udp", "dport": strconv.Itoa(int(udp.DstPort))}).Inc() + continue + } + } +} + +func Start(addr string) error { + if err := prometheus.Register(blockedConnections); err != nil { + return err + } + + http.Handle("/metrics", promhttp.Handler()) + + go func() { + log.Infof("Serving /metrics on %s", addr) + if err := http.ListenAndServe(addr, nil); err != nil { + log.Fatalf("Failed to bind metrics server: %v", err) + } + }() + + go gatherMetrics() + + return nil +} diff --git a/npc/namespace.go b/npc/namespace.go new file mode 100644 index 00000000..b2ae88d4 --- /dev/null +++ b/npc/namespace.go @@ -0,0 +1,337 @@ +package npc + +import ( + "encoding/json" + + log "github.com/Sirupsen/logrus" + "github.com/coreos/go-iptables/iptables" + "k8s.io/client-go/pkg/api/unversioned" + coreapi "k8s.io/client-go/pkg/api/v1" + extnapi "k8s.io/client-go/pkg/apis/extensions/v1beta1" + "k8s.io/client-go/pkg/types" + "k8s.io/client-go/pkg/util/uuid" + + "github.com/weaveworks/weave/npc/ipset" +) + +type ns struct { + ipt *iptables.IPTables // interface to iptables + ips ipset.Interface // interface to ipset + + name string // k8s Namespace name + namespace *coreapi.Namespace // k8s Namespace object + pods map[types.UID]*coreapi.Pod // k8s Pod objects by UID + policies map[types.UID]*extnapi.NetworkPolicy // k8s NetworkPolicy objects by UID + + uid types.UID // surrogate UID to own allPods selector + allPods *selectorSpec // hash:ip ipset of all pod IPs in this namespace + + nsSelectors *selectorSet + podSelectors *selectorSet + rules *ruleSet +} + +func newNS(name string, ipt *iptables.IPTables, ips ipset.Interface, nsSelectors *selectorSet) (*ns, error) { + allPods, err := newSelectorSpec(&unversioned.LabelSelector{}, name, ipset.HashIP) + if err != nil { + return nil, err + } + + ns := &ns{ + ipt: ipt, + ips: ips, + name: name, + pods: make(map[types.UID]*coreapi.Pod), + policies: make(map[types.UID]*extnapi.NetworkPolicy), + uid: uuid.NewUUID(), + allPods: allPods, + nsSelectors: nsSelectors, + rules: newRuleSet(ipt)} + + ns.podSelectors = newSelectorSet(ips, ns.onNewPodSelector) + + if err := ns.podSelectors.provision(ns.uid, nil, map[string]*selectorSpec{ns.allPods.key: ns.allPods}); err != nil { + return nil, err + } + + return ns, nil +} + +func (ns *ns) empty() bool { + return len(ns.pods) == 0 && len(ns.policies) == 0 && ns.namespace == nil +} + +func (ns *ns) destroy() error { + if err := ns.podSelectors.deprovision(ns.uid, map[string]*selectorSpec{ns.allPods.key: ns.allPods}, nil); err != nil { + return err + } + return nil +} + +func (ns *ns) onNewPodSelector(selector *selector) error { + for _, pod := range ns.pods { + if hasIP(pod) { + if selector.matches(pod.ObjectMeta.Labels) { + if err := selector.addEntry(pod.Status.PodIP); err != nil { + return err + } + } + } + } + return nil +} + +func (ns *ns) addPod(obj *coreapi.Pod) error { + ns.pods[obj.ObjectMeta.UID] = obj + + if !hasIP(obj) { + return nil + } + + return ns.podSelectors.addToMatching(obj.ObjectMeta.Labels, obj.Status.PodIP) +} + +func (ns *ns) updatePod(oldObj, newObj *coreapi.Pod) error { + delete(ns.pods, oldObj.ObjectMeta.UID) + ns.pods[newObj.ObjectMeta.UID] = newObj + + if !hasIP(oldObj) && !hasIP(newObj) { + return nil + } + + if hasIP(oldObj) && !hasIP(newObj) { + return ns.podSelectors.delFromMatching(oldObj.ObjectMeta.Labels, oldObj.Status.PodIP) + } + + if !hasIP(oldObj) && hasIP(newObj) { + return ns.podSelectors.addToMatching(newObj.ObjectMeta.Labels, newObj.Status.PodIP) + } + + if !equals(oldObj.ObjectMeta.Labels, newObj.ObjectMeta.Labels) || + oldObj.Status.PodIP != newObj.Status.PodIP { + + for _, ps := range ns.podSelectors.entries { + oldMatch := ps.matches(oldObj.ObjectMeta.Labels) + newMatch := ps.matches(newObj.ObjectMeta.Labels) + if oldMatch == newMatch && oldObj.Status.PodIP == newObj.Status.PodIP { + continue + } + if oldMatch { + if err := ps.delEntry(oldObj.Status.PodIP); err != nil { + return err + } + } + if newMatch { + if err := ps.addEntry(newObj.Status.PodIP); err != nil { + return err + } + } + } + } + + return nil +} + +func (ns *ns) deletePod(obj *coreapi.Pod) error { + delete(ns.pods, obj.ObjectMeta.UID) + + if !hasIP(obj) { + return nil + } + + return ns.podSelectors.delFromMatching(obj.ObjectMeta.Labels, obj.Status.PodIP) +} + +func (ns *ns) addNetworkPolicy(obj *extnapi.NetworkPolicy) error { + ns.policies[obj.ObjectMeta.UID] = obj + + // Analyse policy, determine which rules and ipsets are required + rules, nsSelectors, podSelectors, err := ns.analysePolicy(obj) + if err != nil { + return err + } + + // Provision required resources in dependency order + if err := ns.nsSelectors.provision(obj.ObjectMeta.UID, nil, nsSelectors); err != nil { + return err + } + if err := ns.podSelectors.provision(obj.ObjectMeta.UID, nil, podSelectors); err != nil { + return err + } + if err := ns.rules.provision(obj.ObjectMeta.UID, nil, rules); err != nil { + return err + } + + return nil +} + +func (ns *ns) updateNetworkPolicy(oldObj, newObj *extnapi.NetworkPolicy) error { + delete(ns.policies, oldObj.ObjectMeta.UID) + ns.policies[newObj.ObjectMeta.UID] = newObj + + // Analyse the old and the new policy so we can determine differences + oldRules, oldNsSelectors, oldPodSelectors, err := ns.analysePolicy(oldObj) + if err != nil { + return err + } + newRules, newNsSelectors, newPodSelectors, err := ns.analysePolicy(newObj) + if err != nil { + return err + } + + // Deprovision unused and provision newly required resources in dependency order + if err := ns.rules.deprovision(oldObj.ObjectMeta.UID, oldRules, newRules); err != nil { + return err + } + if err := ns.nsSelectors.deprovision(oldObj.ObjectMeta.UID, oldNsSelectors, newNsSelectors); err != nil { + return err + } + if err := ns.podSelectors.deprovision(oldObj.ObjectMeta.UID, oldPodSelectors, newPodSelectors); err != nil { + return err + } + if err := ns.nsSelectors.provision(oldObj.ObjectMeta.UID, oldNsSelectors, newNsSelectors); err != nil { + return err + } + if err := ns.podSelectors.provision(oldObj.ObjectMeta.UID, oldPodSelectors, newNsSelectors); err != nil { + return err + } + if err := ns.rules.provision(oldObj.ObjectMeta.UID, oldRules, newRules); err != nil { + return err + } + + return nil +} + +func (ns *ns) deleteNetworkPolicy(obj *extnapi.NetworkPolicy) error { + delete(ns.policies, obj.ObjectMeta.UID) + + // Analyse network policy to free resources + rules, nsSelectors, podSelectors, err := ns.analysePolicy(obj) + if err != nil { + return err + } + + // Deprovision unused resources in dependency order + if err := ns.rules.deprovision(obj.ObjectMeta.UID, rules, nil); err != nil { + return err + } + if err := ns.nsSelectors.deprovision(obj.ObjectMeta.UID, nsSelectors, nil); err != nil { + return err + } + if err := ns.podSelectors.deprovision(obj.ObjectMeta.UID, podSelectors, nil); err != nil { + return err + } + + return nil +} + +func bypassRule(nsIpsetName ipset.Name) []string { + return []string{"-m", "set", "--match-set", string(nsIpsetName), "dst", "-j", "ACCEPT"} +} + +func (ns *ns) ensureBypassRule(nsIpsetName ipset.Name) error { + return ns.ipt.Append(TableFilter, DefaultChain, bypassRule(ns.allPods.ipsetName)...) +} + +func (ns *ns) deleteBypassRule(nsIpsetName ipset.Name) error { + return ns.ipt.Delete(TableFilter, DefaultChain, bypassRule(ns.allPods.ipsetName)...) +} + +func (ns *ns) addNamespace(obj *coreapi.Namespace) error { + ns.namespace = obj + + // Insert a rule to bypass policies if namespace is DefaultAllow + if !isDefaultDeny(obj) { + return ns.ensureBypassRule(ns.allPods.ipsetName) + } + + // Add namespace ipset to matching namespace selectors + return ns.nsSelectors.addToMatching(obj.ObjectMeta.Labels, string(ns.allPods.ipsetName)) +} + +func (ns *ns) updateNamespace(oldObj, newObj *coreapi.Namespace) error { + ns.namespace = newObj + + // Update bypass rule if ingress default has changed + oldDefaultDeny := isDefaultDeny(oldObj) + newDefaultDeny := isDefaultDeny(newObj) + + if oldDefaultDeny != newDefaultDeny { + if oldDefaultDeny { + return ns.ensureBypassRule(ns.allPods.ipsetName) + } + if newDefaultDeny { + return ns.deleteBypassRule(ns.allPods.ipsetName) + } + } + + // Re-evaluate namespace selector membership if labels have changed + if !equals(oldObj.ObjectMeta.Labels, newObj.ObjectMeta.Labels) { + for _, selector := range ns.nsSelectors.entries { + oldMatch := selector.matches(oldObj.ObjectMeta.Labels) + newMatch := selector.matches(newObj.ObjectMeta.Labels) + if oldMatch == newMatch { + continue + } + if oldMatch { + if err := selector.delEntry(string(ns.allPods.ipsetName)); err != nil { + return err + } + } + if newMatch { + if err := selector.addEntry(string(ns.allPods.ipsetName)); err != nil { + return err + } + } + } + } + + return nil +} + +func (ns *ns) deleteNamespace(obj *coreapi.Namespace) error { + ns.namespace = nil + + // Remove bypass rule + if !isDefaultDeny(obj) { + return ns.deleteBypassRule(ns.allPods.ipsetName) + } + + // Remove namespace ipset from any matching namespace selectors + return ns.nsSelectors.delFromMatching(obj.ObjectMeta.Labels, string(ns.allPods.ipsetName)) +} + +func hasIP(pod *coreapi.Pod) bool { + // Ensure pod has an IP address and isn't sharing the host network namespace + return len(pod.Status.PodIP) > 0 && !pod.Spec.HostNetwork +} + +func equals(a, b map[string]string) bool { + if len(a) != len(b) { + return false + } + for ak, av := range a { + if b[ak] != av { + return false + } + } + return true +} + +func isDefaultDeny(namespace *coreapi.Namespace) bool { + nnpJson, found := namespace.ObjectMeta.Annotations["net.beta.kubernetes.io/network-policy"] + if !found { + return false + } + + var nnp NamespaceNetworkPolicy + if err := json.Unmarshal([]byte(nnpJson), &nnp); err != nil { + log.Warn("Ignoring network policy annotation: unmarshal failed:", err) + // If we can't understand the annotation, behave as if it isn't present + return false + } + + return nnp.Ingress != nil && + nnp.Ingress.Isolation != nil && + *(nnp.Ingress.Isolation) == DefaultDeny +} diff --git a/npc/rule.go b/npc/rule.go new file mode 100644 index 00000000..0f1308ac --- /dev/null +++ b/npc/rule.go @@ -0,0 +1,77 @@ +package npc + +import ( + "strings" + + log "github.com/Sirupsen/logrus" + "github.com/coreos/go-iptables/iptables" + "k8s.io/client-go/pkg/types" +) + +type ruleSpec struct { + key string + args []string +} + +func newRuleSpec(proto *string, srcHost *selectorSpec, dstHost *selectorSpec, dstPort *string) *ruleSpec { + args := []string{} + if proto != nil { + args = append(args, "-p", *proto) + } + if srcHost != nil { + args = append(args, "-m", "set", "--match-set", string(srcHost.ipsetName), "src") + } + if dstHost != nil { + args = append(args, "-m", "set", "--match-set", string(dstHost.ipsetName), "dst") + } + if dstPort != nil { + args = append(args, "--dport", *dstPort) + } + args = append(args, "-j", "ACCEPT") + key := strings.Join(args, " ") + + return &ruleSpec{key, args} +} + +type ruleSet struct { + ipt *iptables.IPTables + users map[string]map[types.UID]struct{} +} + +func newRuleSet(ipt *iptables.IPTables) *ruleSet { + return &ruleSet{ipt, make(map[string]map[types.UID]struct{})} +} + +func (rs *ruleSet) deprovision(user types.UID, current, desired map[string]*ruleSpec) error { + for key, spec := range current { + if _, found := desired[key]; !found { + delete(rs.users[key], user) + if len(rs.users[key]) == 0 { + log.Infof("deleting rule: %v", spec.args) + if err := rs.ipt.Delete(TableFilter, IngressChain, spec.args...); err != nil { + return err + } + delete(rs.users, key) + } + } + } + + return nil +} + +func (rs *ruleSet) provision(user types.UID, current, desired map[string]*ruleSpec) error { + for key, spec := range desired { + if _, found := current[key]; !found { + if _, found := rs.users[key]; !found { + log.Infof("adding rule: %v", spec.args) + if err := rs.ipt.Append(TableFilter, IngressChain, spec.args...); err != nil { + return err + } + rs.users[key] = make(map[types.UID]struct{}) + } + rs.users[key][user] = struct{}{} + } + } + + return nil +} diff --git a/npc/selector.go b/npc/selector.go new file mode 100644 index 00000000..247051d8 --- /dev/null +++ b/npc/selector.go @@ -0,0 +1,130 @@ +package npc + +import ( + log "github.com/Sirupsen/logrus" + "k8s.io/client-go/pkg/api/unversioned" + "k8s.io/client-go/pkg/labels" + "k8s.io/client-go/pkg/types" + + "github.com/weaveworks/weave/npc/ipset" +) + +type selectorSpec struct { + key string // string representation (for hash keying/equality comparison) + selector labels.Selector // k8s Selector object (for matching) + + ipsetType ipset.Type // type of ipset to provision + ipsetName ipset.Name // generated ipset name +} + +func newSelectorSpec(json *unversioned.LabelSelector, nsName string, ipsetType ipset.Type) (*selectorSpec, error) { + selector, err := unversioned.LabelSelectorAsSelector(json) + if err != nil { + return nil, err + } + key := selector.String() + return &selectorSpec{ + key: key, + selector: selector, + // We prefix the selector string with the namespace name when generating + // the shortname because you can specify the same selector in multiple + // namespaces - we need those to map to distinct ipsets + ipsetName: ipset.Name("weave-" + shortName(nsName+":"+key)), + ipsetType: ipsetType}, nil +} + +type selector struct { + ips ipset.Interface + spec *selectorSpec +} + +func (s *selector) matches(labelMap map[string]string) bool { + return s.spec.selector.Matches(labels.Set(labelMap)) +} + +func (s *selector) addEntry(entry string) error { + 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 { + log.Infof("deleting entry %s from %s", entry, s.spec.ipsetName) + return s.ips.DelEntry(s.spec.ipsetName, entry) +} + +type selectorFn func(selector *selector) error + +type selectorSet struct { + ips ipset.Interface + onNewSelector selectorFn + users map[string]map[types.UID]struct{} // list of users per selector + entries map[string]*selector +} + +func newSelectorSet(ips ipset.Interface, onNewSelector selectorFn) *selectorSet { + return &selectorSet{ + ips: ips, + onNewSelector: onNewSelector, + users: make(map[string]map[types.UID]struct{}), + entries: make(map[string]*selector)} +} + +func (ss *selectorSet) addToMatching(labelMap map[string]string, entry string) error { + for _, s := range ss.entries { + if s.matches(labelMap) { + if err := s.addEntry(entry); err != nil { + return err + } + } + } + return nil +} + +func (ss *selectorSet) delFromMatching(labelMap map[string]string, entry string) error { + for _, s := range ss.entries { + if s.matches(labelMap) { + if err := s.delEntry(entry); err != nil { + return err + } + } + } + return nil +} + +func (ss *selectorSet) deprovision(user types.UID, current, desired map[string]*selectorSpec) error { + for key, spec := range current { + if _, found := desired[key]; !found { + delete(ss.users[key], user) + if len(ss.users[key]) == 0 { + log.Infof("destroying ipset: %#v", spec) + if err := ss.ips.Destroy(spec.ipsetName); err != nil { + return err + } + delete(ss.entries, key) + delete(ss.users, key) + } + } + } + return nil +} + +func (ss *selectorSet) provision(user types.UID, current, desired map[string]*selectorSpec) error { + for key, spec := range desired { + if _, found := current[key]; !found { + if _, found := ss.users[key]; !found { + log.Infof("creating ipset: %#v", spec) + if err := ss.ips.Create(spec.ipsetName, spec.ipsetType); err != nil { + return err + } + selector := &selector{ss.ips, spec} + if err := ss.onNewSelector(selector); err != nil { + return err + } + ss.users[key] = make(map[types.UID]struct{}) + ss.entries[key] = selector + } + ss.users[key][user] = struct{}{} + } + } + return nil +} diff --git a/npc/shortname.go b/npc/shortname.go new file mode 100644 index 00000000..c0704bfb --- /dev/null +++ b/npc/shortname.go @@ -0,0 +1,36 @@ +package npc + +import ( + "crypto/sha1" + "math/big" +) + +const ( + // This array: + // * Must include only printable UTF8 characters that are represented with a single byte + // * Must be at least of length 85 (`len("weave-") + l(2^160)/l(85)` equals 31, the maximum ipset name length) + // * Must not include commas as those are treated specially by `ipset add` when adding a named set to a list:set + // * Should not include space for readability + // * Should not include single quote or backslash to be nice to shell users + ShortNameSymbols = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789(){}[]<>_$%^&*|/?.;:@#~" +) + +// sha1 hash an arbitrary string and represent it using the full range of +// printable ascii characters +func shortName(arbitrary string) string { + symbols := []byte(ShortNameSymbols) + sum := sha1.Sum([]byte(arbitrary)) + i := big.NewInt(0).SetBytes(sum[:]) + base := big.NewInt(int64(len(symbols))) + zero := big.NewInt(0) + result := make([]byte, 0) + + for i.Cmp(zero) > 0 { + remainder := new(big.Int).Mod(i, base) + i.Sub(i, remainder) + i.Div(i, base) + result = append(result, symbols[remainder.Int64()]) + } + + return string(result) +} diff --git a/npc/ulogd/ulogd.go b/npc/ulogd/ulogd.go new file mode 100644 index 00000000..c38fb1e6 --- /dev/null +++ b/npc/ulogd/ulogd.go @@ -0,0 +1,29 @@ +package ulogd + +import ( + log "github.com/Sirupsen/logrus" + "io" + "os" + "os/exec" +) + +func waitForExit(cmd *exec.Cmd) { + if err := cmd.Wait(); err != nil { + log.Fatalf("ulogd terminated: %v", err) + } + log.Fatal("ulogd terminated normally") +} + +func Start() error { + cmd := exec.Command("/usr/sbin/ulogd", "-v") + stdout, err := cmd.StderrPipe() + if err != nil { + return err + } + if err := cmd.Start(); err != nil { + return err + } + go io.Copy(os.Stdout, stdout) + go waitForExit(cmd) + return nil +} diff --git a/prog/kube-peers/main.go b/prog/kube-peers/main.go index 8759300a..977a4872 100644 --- a/prog/kube-peers/main.go +++ b/prog/kube-peers/main.go @@ -4,9 +4,9 @@ import ( "fmt" "log" - "k8s.io/client-go/1.4/kubernetes" - "k8s.io/client-go/1.4/pkg/api" - "k8s.io/client-go/1.4/rest" + "k8s.io/client-go/kubernetes" + api "k8s.io/client-go/pkg/api/v1" + "k8s.io/client-go/rest" ) func getKubePeers() ([]string, error) { diff --git a/prog/weave-npc/Dockerfile b/prog/weave-npc/Dockerfile new file mode 100644 index 00000000..4e011ac5 --- /dev/null +++ b/prog/weave-npc/Dockerfile @@ -0,0 +1,10 @@ +FROM alpine +RUN apk add --update \ + iptables \ + ipset \ + ulogd \ + && rm -rf /var/cache/apk/* \ + && mknod /var/log/ulogd.pcap p +COPY ./weave-npc /usr/bin/weave-npc +COPY ./ulogd.conf /etc/ulogd.conf +ENTRYPOINT ["/usr/bin/weave-npc"] diff --git a/prog/weave-npc/main.go b/prog/weave-npc/main.go new file mode 100644 index 00000000..bdaee205 --- /dev/null +++ b/prog/weave-npc/main.go @@ -0,0 +1,180 @@ +package main + +import ( + "os" + "os/signal" + "syscall" + + log "github.com/Sirupsen/logrus" + "github.com/coreos/go-iptables/iptables" + "github.com/spf13/cobra" + "k8s.io/client-go/kubernetes" + coreapi "k8s.io/client-go/pkg/api/v1" + extnapi "k8s.io/client-go/pkg/apis/extensions/v1beta1" + "k8s.io/client-go/pkg/fields" + "k8s.io/client-go/pkg/runtime" + "k8s.io/client-go/pkg/util/wait" + "k8s.io/client-go/rest" + "k8s.io/client-go/tools/cache" + + "github.com/weaveworks/weave/npc" + "github.com/weaveworks/weave/npc/ipset" + "github.com/weaveworks/weave/npc/metrics" + "github.com/weaveworks/weave/npc/ulogd" +) + +var ( + version = "(unreleased)" + metricsAddr string +) + +func handleError(err error) { + if err != nil { + log.Fatal(err) + } +} + +func makeController(getter cache.Getter, resource string, + objType runtime.Object, handlers cache.ResourceEventHandlerFuncs) *cache.Controller { + listWatch := cache.NewListWatchFromClient(getter, resource, "", fields.Everything()) + _, controller := cache.NewInformer(listWatch, objType, 0, handlers) + return controller +} + +func resetIPTables(ipt *iptables.IPTables) error { + // Flush chains first so there are no refs to extant ipsets + if err := ipt.ClearChain(npc.TableFilter, npc.IngressChain); err != nil { + return err + } + + if err := ipt.ClearChain(npc.TableFilter, npc.DefaultChain); err != nil { + return err + } + + if err := ipt.ClearChain(npc.TableFilter, npc.MainChain); err != nil { + 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 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 +} + +func resetIPSets(ips ipset.Interface) error { + // TODO should restrict ipset operations to the `weave-` prefix: + + if err := ips.FlushAll(); err != nil { + return err + } + + if err := ips.DestroyAll(); err != nil { + return err + } + + return nil +} + +func root(cmd *cobra.Command, args []string) { + log.Infof("Starting Weaveworks NPC %s", version) + + if err := metrics.Start(metricsAddr); err != nil { + log.Fatalf("Failed to start metrics: %v", err) + } + + if err := ulogd.Start(); err != nil { + log.Fatalf("Failed to start ulogd: %v", err) + } + + config, err := rest.InClusterConfig() + if err != nil { + log.Fatal(err) + } + + client, err := kubernetes.NewForConfig(config) + if err != nil { + log.Fatal(err) + } + + ipt, err := iptables.New() + if err != nil { + log.Fatal(err) + } + + ips := ipset.New() + + handleError(resetIPTables(ipt)) + handleError(resetIPSets(ips)) + + npc := npc.New(ipt, ips) + + nsController := makeController(client.Core().RESTClient(), "namespaces", &coreapi.Namespace{}, + cache.ResourceEventHandlerFuncs{ + AddFunc: func(obj interface{}) { + handleError(npc.AddNamespace(obj.(*coreapi.Namespace))) + }, + DeleteFunc: func(obj interface{}) { + handleError(npc.DeleteNamespace(obj.(*coreapi.Namespace))) + }, + UpdateFunc: func(old, new interface{}) { + handleError(npc.UpdateNamespace(old.(*coreapi.Namespace), new.(*coreapi.Namespace))) + }}) + + podController := makeController(client.Core().RESTClient(), "pods", &coreapi.Pod{}, + cache.ResourceEventHandlerFuncs{ + AddFunc: func(obj interface{}) { + handleError(npc.AddPod(obj.(*coreapi.Pod))) + }, + DeleteFunc: func(obj interface{}) { + handleError(npc.DeletePod(obj.(*coreapi.Pod))) + }, + UpdateFunc: func(old, new interface{}) { + handleError(npc.UpdatePod(old.(*coreapi.Pod), new.(*coreapi.Pod))) + }}) + + npController := makeController(client.Extensions().RESTClient(), "networkpolicies", &extnapi.NetworkPolicy{}, + cache.ResourceEventHandlerFuncs{ + AddFunc: func(obj interface{}) { + handleError(npc.AddNetworkPolicy(obj.(*extnapi.NetworkPolicy))) + }, + DeleteFunc: func(obj interface{}) { + handleError(npc.DeleteNetworkPolicy(obj.(*extnapi.NetworkPolicy))) + }, + UpdateFunc: func(old, new interface{}) { + handleError(npc.UpdateNetworkPolicy(old.(*extnapi.NetworkPolicy), new.(*extnapi.NetworkPolicy))) + }}) + + go nsController.Run(wait.NeverStop) + go podController.Run(wait.NeverStop) + go npController.Run(wait.NeverStop) + + signals := make(chan os.Signal, 1) + signal.Notify(signals, syscall.SIGINT, syscall.SIGTERM) + log.Fatalf("Exiting: %v", <-signals) +} + +func main() { + rootCmd := &cobra.Command{ + Use: "weave-npc", + Short: "Weaveworks Kubernetes Network Policy Controller", + Run: root} + + rootCmd.PersistentFlags().StringVar(&metricsAddr, "metrics-addr", ":8686", "metrics server bind address") + + if err := rootCmd.Execute(); err != nil { + log.Fatal(err) + } +} diff --git a/prog/weave-npc/ulogd.conf b/prog/weave-npc/ulogd.conf new file mode 100644 index 00000000..7c358bf2 --- /dev/null +++ b/prog/weave-npc/ulogd.conf @@ -0,0 +1,13 @@ +[global] +logfile="/dev/null" +plugin="/usr/lib/ulogd/ulogd_inppkt_NFLOG.so" +plugin="/usr/lib/ulogd/ulogd_raw2packet_BASE.so" +plugin="/usr/lib/ulogd/ulogd_output_PCAP.so" +stack=log1:NFLOG,base1:BASE,pcap1:PCAP + +[log1] +group=86 + +[pcap1] +file="/var/log/ulogd.pcap" +sync=1 diff --git a/vendor/github.com/pkg/errors b/vendor/github.com/pkg/errors new file mode 160000 index 00000000..248dadf4 --- /dev/null +++ b/vendor/github.com/pkg/errors @@ -0,0 +1 @@ +Subproject commit 248dadf4e9068a0b3e79f02ed0a610d935de5302 diff --git a/vendor/github.com/spf13/cobra b/vendor/github.com/spf13/cobra new file mode 160000 index 00000000..6e91dded --- /dev/null +++ b/vendor/github.com/spf13/cobra @@ -0,0 +1 @@ +Subproject commit 6e91dded25d73176bf7f60b40dd7aa1f0bf9be8d diff --git a/vendor/github.com/spf13/pflag b/vendor/github.com/spf13/pflag new file mode 160000 index 00000000..5ccb023b --- /dev/null +++ b/vendor/github.com/spf13/pflag @@ -0,0 +1 @@ +Subproject commit 5ccb023bc27df288a957c5e994cd44fd19619465 diff --git a/vendor/github.com/spf13/plag b/vendor/github.com/spf13/plag new file mode 160000 index 00000000..5ccb023b --- /dev/null +++ b/vendor/github.com/spf13/plag @@ -0,0 +1 @@ +Subproject commit 5ccb023bc27df288a957c5e994cd44fd19619465 diff --git a/vendor/k8s.io/client-go b/vendor/k8s.io/client-go index d72c0e16..b22087a5 160000 --- a/vendor/k8s.io/client-go +++ b/vendor/k8s.io/client-go @@ -1 +1 @@ -Subproject commit d72c0e162789e1bbb33c33cfa26858a1375efe01 +Subproject commit b22087a53becae45931ed72d5e0f12e0031d771a