cni: lock each pod's port to the pod's own addresses

A pod could put any source MAC or address on the wire: weave has no layer-2
anti-spoofing, so a pod with NET_RAW could pose as another pod or as a node's
bridge, on its own node (local bridging never reaches the router) and, through
weave, on others. That undercut every rule that trusts a source MAC or address,
including the egress holder's check of which node a tunnel packet came from.

The CNI plugin now locks the host side of each pod's veth with tc ingress
filters: IPv4 from the pod's MAC and address, ARP speaking for exactly those
(or probing from 0.0.0.0) pass; everything else is dropped before the bridge
sees it. The u32 keys follow iproute2's layout for "ether src" (-8, -4) and
"ip src" (12). ADD fails if the port cannot be locked, so an unlocked pod never
starts. The filters go away with the veth.

Pods attached before this plugin get locked by weave-init: "weaveutil
lock-ports" walks libcni's result cache (each attachment's netns, interface and
address) and locks their ports, leaving locked ones alone. weave-init mounts
the cache and /var/run/netns read-only for it.

TestPortLockInKernel (root) builds a bridge and two pods in throwaway
namespaces and checks what reaches the other pod; reweave/test/portlock runs
the same against a live cluster, across nodes, after checking every port is
locked.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UHXzGrsxfkGk8LmwrBbHNZ

.gitea/workflows/images.yml

@@ -54,3 +54,14 @@ jobs:
               .
             echo "pushed $image"
           done
+
+      # The images live in the registry now; the runner keeps only a bounded build cache. Without this the
+      # runner's 10Gi Docker volume fills up after a handful of builds (no space left on device mid-build).
+      - name: Free the runner's disk
+        if: always()
+        run: |
+          set -u
+          docker image ls --format '{{.Repository}}:{{.Tag}}' | grep "^$IMAGE_PREFIX/" | xargs -r docker image rm -f
+          docker image prune -f
+          docker builder prune -f --keep-storage 4GB
+          docker system df

lb/admit/admit.go

@@ -0,0 +1,88 @@
+// Package admit is the protocol between the weave CNI plugin and weave-npc's egress agent: at CNI ADD the
+// plugin tells the agent which pod got which IP, and the agent answers once the pod's egress rules are in
+// place, so an EgressService member's first packet already leaves from the Service's address (or not at all).
+//
+// It depends on the standard library only: the plugin binary must not grow a Kubernetes client.
+package admit
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"strings"
+	"time"
+)
+
+// SocketPath is where the agent listens (a hostPath directory shared with the host, where the plugin runs).
+const SocketPath = "/run/weave-npc/egress.sock"
+
+// Request is one pod's admission: the IP the plugin just allocated to it.
+type Request struct {
+	Namespace string `json:"namespace"`
+	Name      string `json:"name"`
+	IP        string `json:"ip"`
+}
+
+// Response says whether the pod is an EgressService member (its rules are installed either way when 200).
+type Response struct {
+	Member bool `json:"member"`
+}
+
+// Timeout bounds one admission (the agent may have to fetch the pod from the API server).
+const Timeout = 20 * time.Second
+
+// PodFromArgs reads the pod's namespace and name from CNI_ARGS ("K8S_POD_NAMESPACE=..;K8S_POD_NAME=..").
+func PodFromArgs(args string) (namespace, name string) {
+	for _, kv := range strings.Split(args, ";") {
+		k, v, ok := strings.Cut(kv, "=")
+		if !ok {
+			continue
+		}
+		switch k {
+		case "K8S_POD_NAMESPACE":
+			namespace = v
+		case "K8S_POD_NAME":
+			name = v
+		}
+	}
+	return namespace, name
+}
+
+// Admit asks the agent at socket to admit req, and returns only once the agent has installed the pod's rules.
+func Admit(ctx context.Context, socket string, req Request) (Response, error) {
+	var resp Response
+	body, err := json.Marshal(req)
+	if err != nil {
+		return resp, err
+	}
+	client := &http.Client{Transport: &http.Transport{
+		DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
+			var d net.Dialer
+			return d.DialContext(ctx, "unix", socket)
+		},
+	}}
+	ctx, cancel := context.WithTimeout(ctx, Timeout)
+	defer cancel()
+	hreq, err := http.NewRequestWithContext(ctx, http.MethodPost, "http://weave-npc/admit", bytes.NewReader(body))
+	if err != nil {
+		return resp, err
+	}
+	hreq.Header.Set("Content-Type", "application/json")
+	hresp, err := client.Do(hreq)
+	if err != nil {
+		return resp, fmt.Errorf("egress admission (weave-npc at %s): %v", socket, err)
+	}
+	defer hresp.Body.Close()
+	out, _ := io.ReadAll(io.LimitReader(hresp.Body, 1<<16))
+	if hresp.StatusCode != http.StatusOK {
+		return resp, fmt.Errorf("egress admission refused (%s): %s", hresp.Status, strings.TrimSpace(string(out)))
+	}
+	if err := json.Unmarshal(out, &resp); err != nil {
+		return resp, fmt.Errorf("egress admission: bad answer: %v", err)
+	}
+	return resp, nil
+}

lb/controller.go

@@ -4,42 +4,66 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"net"
+	"net/http"
+	"os"
+	"path/filepath"
+	"sync"
 	"time"
 
 	coreapi "k8s.io/api/core/v1"
 	discovery "k8s.io/api/discovery/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/cache"
 
 	"github.com/rajch/weave/common"
+	"github.com/rajch/weave/lb/admit"
 )
 
-// Controller watches Services, EndpointSlices and Nodes and keeps this node's part of the picture true.
+// Controller watches Services, EndpointSlices, Nodes, EgressServices and this node's Pods, and keeps this
+// node's part of the picture true. It also answers the CNI plugin's egress admissions.
 type Controller struct {
 	self   string
 	client kubernetes.Interface
+	egress rest.Interface
 	dp     *Datapath
 	// readLocal is the node's own interfaces (swappable in tests)
 	readLocal func() (Local, error)
 
-	services, slices, nodes cache.Store
-	kick                    chan struct{}
-	lastProblems            map[string]bool
+	services, slices, nodes, pods, egresses cache.Store
+	synced                                  []cache.InformerSynced
+	egressSynced                            cache.InformerSynced // nil while the EgressService API is absent
+	kick                                    chan struct{}
+	ready                                   chan struct{} // closed once the first pass ran
+
+	mu           sync.Mutex // one pass at a time (the loop, or an admission)
+	lastProblems map[string]bool
 	// Services whose status this node wrote since it started (so a restart rewrites ipMode once)
 	written map[types.UID]string
+	// admitted: IP -> pod the CNI plugin gave it; fetched: pods admitted before the pod informer had them
+	admitted map[string]Admission
+	fetched  map[string]*coreapi.Pod
+	at       map[string]time.Time
+	// lastEgress is what was last applied, reused while the EgressService caches are not synced (never wipe
+	// members' rules on a partial view)
+	lastEgress EgressDesired
 }
 
 // Resync is how often the node re-reads its interfaces and re-applies everything even without API events.
 const Resync = 30 * time.Second
 
-// NewController runs nothing yet; Run does.
-func NewController(self string, client kubernetes.Interface, dp *Datapath) *Controller {
-	return &Controller{self: self, client: client, dp: dp, readLocal: ReadLocal,
-		kick: make(chan struct{}, 1), lastProblems: map[string]bool{}, written: map[types.UID]string{}}
+// NewController runs nothing yet; Run does. egress is the EgressService API client.
+func NewController(self string, client kubernetes.Interface, egress rest.Interface, dp *Datapath) *Controller {
+	return &Controller{self: self, client: client, egress: egress, dp: dp, readLocal: ReadLocal,
+		kick: make(chan struct{}, 1), ready: make(chan struct{}), lastProblems: map[string]bool{},
+		written: map[types.UID]string{}, admitted: map[string]Admission{}, fetched: map[string]*coreapi.Pod{},
+		at: map[string]time.Time{}}
 }
 
 func (c *Controller) poke() {
@@ -49,26 +73,84 @@ func (c *Controller) poke() {
 	}
 }
 
+func newInformer(c *Controller, lw cache.ListerWatcher, obj runtime.Object) (cache.Store, cache.Controller) {
+	return cache.NewInformer(lw, obj, 0, cache.ResourceEventHandlerFuncs{
+		AddFunc:    func(interface{}) { c.poke() },
+		UpdateFunc: func(interface{}, interface{}) { c.poke() },
+		DeleteFunc: func(interface{}) { c.poke() },
+	})
+}
+
+// egressAPI reports whether the EgressService API is served (the CRD is installed).
+func (c *Controller) egressAPI() (bool, error) {
+	_, err := c.client.Discovery().ServerResourcesForGroupVersion(GroupVersion.String())
+	if err == nil {
+		return true, nil
+	}
+	if apierrors.IsNotFound(err) {
+		return false, nil
+	}
+	return false, err
+}
+
+func (c *Controller) startEgress(ctx context.Context) {
+	var ctl cache.Controller
+	c.egresses, ctl = newInformer(c, cache.NewListWatchFromClient(c.egress, EgressServices, "", fields.Everything()), &EgressService{})
+	go ctl.Run(ctx.Done())
+	c.egressSynced = ctl.HasSynced
+}
+
 // Run blocks until ctx ends.
 func (c *Controller) Run(ctx context.Context) {
-	var sc, ec, nc cache.Controller
-	c.services, sc = newInformer(c, c.client.CoreV1().RESTClient(), "services", &coreapi.Service{})
-	c.slices, ec = newInformer(c, c.client.DiscoveryV1().RESTClient(), "endpointslices", &discovery.EndpointSlice{})
-	c.nodes, nc = newInformer(c, c.client.CoreV1().RESTClient(), "nodes", &coreapi.Node{})
-	for _, ctl := range []cache.Controller{sc, ec, nc} {
+	var sc, ec, nc, pc cache.Controller
+	c.services, sc = newInformer(c, cache.NewListWatchFromClient(c.client.CoreV1().RESTClient(), "services", "", fields.Everything()), &coreapi.Service{})
+	c.slices, ec = newInformer(c, cache.NewListWatchFromClient(c.client.DiscoveryV1().RESTClient(), "endpointslices", "", fields.Everything()), &discovery.EndpointSlice{})
+	c.nodes, nc = newInformer(c, cache.NewListWatchFromClient(c.client.CoreV1().RESTClient(), "nodes", "", fields.Everything()), &coreapi.Node{})
+	c.pods, pc = newInformer(c, cache.NewListWatchFromClient(c.client.CoreV1().RESTClient(), "pods", "",
+		fields.OneTermEqualSelector("spec.nodeName", c.self)), &coreapi.Pod{})
+	for _, ctl := range []cache.Controller{sc, ec, nc, pc} {
 		go ctl.Run(ctx.Done())
+		c.synced = append(c.synced, ctl.HasSynced)
 	}
-	if !cache.WaitForCacheSync(ctx.Done(), sc.HasSynced, ec.HasSynced, nc.HasSynced) {
+	// with the API present, nothing is applied before EgressServices are known: an empty view would wipe
+	// members' rules (their traffic would leave from the node); without it, there can be no members
+	for {
+		present, err := c.egressAPI()
+		if err == nil {
+			if present {
+				c.startEgress(ctx)
+				c.synced = append(c.synced, c.egressSynced)
+			}
+			break
+		}
+		common.Log.Warnf("lb: cannot tell whether the EgressService API is served (retrying): %v", err)
+		select {
+		case <-ctx.Done():
+			return
+		case <-time.After(5 * time.Second):
+		}
+	}
+	go c.serveAdmission(ctx)
+	if !cache.WaitForCacheSync(ctx.Done(), c.synced...) {
 		return
 	}
 	tick := time.NewTicker(Resync)
 	defer tick.Stop()
 	c.poke()
+	first := true
 	for {
 		select {
 		case <-ctx.Done():
 			return
 		case <-tick.C:
+			if c.egressSynced == nil {
+				// the API may have been installed since
+				if present, err := c.egressAPI(); err == nil && present {
+					c.mu.Lock()
+					c.startEgress(ctx)
+					c.mu.Unlock()
+				}
+			}
 		case <-c.kick:
 			// let a burst of events settle into one pass
 			time.Sleep(200 * time.Millisecond)
@@ -77,22 +159,25 @@ func (c *Controller) Run(ctx context.Context) {
 			default:
 			}
 		}
-		if err := c.sync(ctx); err != nil {
+		c.mu.Lock()
+		err := c.sync(ctx)
+		c.mu.Unlock()
+		if err != nil {
 			common.Log.Errorf("lb: %v", err)
 		}
+		if first {
+			first = false
+			close(c.ready)
+		}
 	}
 }
 
-func newInformer(c *Controller, getter cache.Getter, resource string, obj runtime.Object) (cache.Store, cache.Controller) {
-	lw := cache.NewListWatchFromClient(getter, resource, "", fields.Everything())
-	return cache.NewInformer(lw, obj, 0, cache.ResourceEventHandlerFuncs{
-		AddFunc:    func(interface{}) { c.poke() },
-		UpdateFunc: func(interface{}, interface{}) { c.poke() },
-		DeleteFunc: func(interface{}) { c.poke() },
-	})
+// egressViewSynced: the EgressService view can be trusted (absent API: trivially empty; else synced).
+func (c *Controller) egressViewSynced() bool {
+	return c.egressSynced == nil || c.egressSynced()
 }
 
-// sync is one pass: publish this node, compute, apply, write statuses.
+// sync is one pass: publish this node, compute, apply, write statuses. Callers hold c.mu.
 func (c *Controller) sync(ctx context.Context) error {
 	local, err := c.readLocal()
 	if err != nil {
@@ -129,8 +214,17 @@ func (c *Controller) sync(ctx context.Context) error {
 			d.Status = append(d.Status, Status{Namespace: f.Namespace, Name: f.Name, Address: f.Address})
 		}
 	}
+
+	e := c.lastEgress
+	e.Status = nil // reusing the datapath, never the writes of an earlier view
+	if c.egressViewSynced() {
+		e = c.computeEgress(nodes, svcs)
+	} else {
+		problems = append(problems, "EgressServices not synced yet: keeping the egress rules last applied")
+	}
+
 	current := map[string]bool{}
-	for _, p := range append(problems, d.Problems...) {
+	for _, p := range append(append(problems, d.Problems...), e.Problems...) {
 		current[p] = true
 		if !c.lastProblems[p] {
 			common.Log.Warnf("lb: %s", p)
@@ -138,17 +232,67 @@ func (c *Controller) sync(ctx context.Context) error {
 	}
 	c.lastProblems = current
 
-	if err := c.dp.Apply(d, local.PodNet); err != nil {
+	if err := c.dp.Apply(d, e, c.self, local); err != nil {
 		return fmt.Errorf("applying datapath: %v", err)
 	}
+	c.lastEgress = e
 	for _, st := range d.Status {
 		if err := c.writeStatus(ctx, st); err != nil {
 			return err
 		}
 	}
+	for _, st := range e.Status {
+		if err := PatchEgressStatus(ctx, c.egress, st.Namespace, st.Name, st.Status); err != nil {
+			return fmt.Errorf("status of egress %s/%s: %v", st.Namespace, st.Name, err)
+		}
+		common.Log.Infof("lb: egress %s/%s: host %q, address %q, %d port blocks", st.Namespace, st.Name, st.Status.Host, st.Status.Address, len(st.Status.Blocks))
+	}
 	return nil
 }
 
+// computeEgress gathers this node's pods (informer, plus pods admitted before the informer had them) and computes.
+func (c *Controller) computeEgress(nodes []Node, svcs []*coreapi.Service) EgressDesired {
+	var egresses []*EgressService
+	if c.egresses != nil {
+		for _, o := range c.egresses.List() {
+			egresses = append(egresses, o.(*EgressService))
+		}
+	}
+	seen := map[string]bool{}
+	var pods []*coreapi.Pod
+	for _, o := range c.pods.List() {
+		p := o.(*coreapi.Pod)
+		seen[p.Namespace+"/"+p.Name] = true
+		pods = append(pods, p)
+	}
+	// forget admissions the API has caught up with or that are long gone
+	for ip, a := range c.admitted {
+		key := a.Namespace + "/" + a.Name
+		o, ok, _ := c.pods.GetByKey(key)
+		caughtUp := ok && o.(*coreapi.Pod).Status.PodIP == ip
+		if caughtUp || (!ok && time.Since(c.at[ip]) > 2*time.Minute) {
+			delete(c.admitted, ip)
+			delete(c.at, ip)
+		}
+	}
+	for key, p := range c.fetched {
+		stillAdmitted := false
+		for _, a := range c.admitted {
+			if a.Namespace+"/"+a.Name == key {
+				stillAdmitted = true
+			}
+		}
+		if !stillAdmitted {
+			delete(c.fetched, key)
+			continue
+		}
+		if !seen[key] {
+			pods = append(pods, p)
+		}
+	}
+	return ComputeEgress(c.self, nodes, svcs, egresses, pods, c.admitted)
+}
+
 func hasStatusFor(d Desired, f Forward) bool {
 	for _, s := range d.Status {
 		if s.Namespace == f.Namespace && s.Name == f.Name {
@@ -166,6 +310,101 @@ func (c *Controller) service(ns, name string) *coreapi.Service {
 	return o.(*coreapi.Service)
 }
 
+// serveAdmission answers the CNI plugin on admit.SocketPath: record the pod's IP, run a pass, answer once the
+// pod's rules are in place. A pod is refused (the plugin fails the ADD, kubelet retries) while this node cannot
+// yet tell whether it is a member.
+func (c *Controller) serveAdmission(ctx context.Context) {
+	if err := os.MkdirAll(filepath.Dir(admit.SocketPath), 0o700); err != nil {
+		common.Log.Errorf("lb: admission socket: %v", err)
+		return
+	}
+	_ = os.Remove(admit.SocketPath)
+	l, err := net.Listen("unix", admit.SocketPath)
+	if err != nil {
+		common.Log.Errorf("lb: admission socket: %v", err)
+		return
+	}
+	_ = os.Chmod(admit.SocketPath, 0o600)
+	mux := http.NewServeMux()
+	mux.HandleFunc("/admit", func(w http.ResponseWriter, r *http.Request) {
+		var req admit.Request
+		if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<16)).Decode(&req); err != nil {
+			http.Error(w, "bad request: "+err.Error(), http.StatusBadRequest)
+			return
+		}
+		member, code, err := c.admit(r.Context(), req)
+		if err != nil {
+			common.Log.Warnf("lb: admission of %s/%s (%s): %v", req.Namespace, req.Name, req.IP, err)
+			http.Error(w, err.Error(), code)
+			return
+		}
+		_ = json.NewEncoder(w).Encode(admit.Response{Member: member})
+	})
+	srv := &http.Server{Handler: mux, ReadHeaderTimeout: 5 * time.Second}
+	go func() {
+		<-ctx.Done()
+		_ = srv.Close()
+	}()
+	if err := srv.Serve(l); err != nil && err != http.ErrServerClosed {
+		common.Log.Errorf("lb: admission server: %v", err)
+	}
+}
+
+func (c *Controller) admit(ctx context.Context, req admit.Request) (bool, int, error) {
+	ip := net.ParseIP(req.IP).To4()
+	if req.Namespace == "" || req.Name == "" || ip == nil {
+		return false, http.StatusBadRequest, fmt.Errorf("need namespace, name and an IPv4 address")
+	}
+	select {
+	case <-c.ready:
+	case <-ctx.Done():
+		return false, http.StatusServiceUnavailable, ctx.Err()
+	case <-time.After(admit.Timeout - 5*time.Second):
+		return false, http.StatusServiceUnavailable, fmt.Errorf("not ready yet")
+	}
+	key := req.Namespace + "/" + req.Name
+	var pod *coreapi.Pod
+	if o, ok, _ := c.pods.GetByKey(key); ok {
+		pod = o.(*coreapi.Pod)
+	} else {
+		gctx, cancel := context.WithTimeout(ctx, 10*time.Second)
+		p, err := c.client.CoreV1().Pods(req.Namespace).Get(gctx, req.Name, metav1.GetOptions{})
+		cancel()
+		if err != nil {
+			return false, http.StatusServiceUnavailable, fmt.Errorf("fetching pod: %v", err)
+		}
+		pod = p
+	}
+	if pod.Spec.NodeName != c.self {
+		return false, http.StatusConflict, fmt.Errorf("pod is bound to %q, not this node", pod.Spec.NodeName)
+	}
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if !c.egressViewSynced() {
+		return false, http.StatusServiceUnavailable, fmt.Errorf("EgressServices not synced yet")
+	}
+	c.admitted[ip.String()] = Admission{Namespace: req.Namespace, Name: req.Name}
+	c.at[ip.String()] = time.Now()
+	if _, ok, _ := c.pods.GetByKey(key); !ok {
+		c.fetched[key] = pod
+	}
+	if err := c.sync(ctx); err != nil {
+		// with no EgressService and no member rules in place there is nothing to protect or to clean up for
+		// this IP: a broken pass must not stop every pod on the node from starting
+		if len(c.lastEgress.Members) == 0 && (c.egresses == nil || len(c.egresses.List()) == 0) {
+			common.Log.Warnf("lb: admitting %s/%s despite a failed pass (no EgressServices): %v", req.Namespace, req.Name, err)
+			return false, http.StatusOK, nil
+		}
+		return false, http.StatusServiceUnavailable, err
+	}
+	for _, m := range c.lastEgress.Members {
+		if m.IP.Equal(ip) {
+			return true, http.StatusOK, nil
+		}
+	}
+	return false, http.StatusOK, nil
+}
+
 // publish keeps this node's annotations equal to what its interfaces say.
 func (c *Controller) publish(ctx context.Context, local Local) error {
 	want := local.Annotations()

lb/datapath.go

@@ -44,17 +44,18 @@ func portChain(f Forward, p Port) string {
 
 // Rules is the full content of the chains this package owns, in iptables-restore form.
 type Rules struct {
-	Nat, Mangle string
+	Nat, Mangle, Filter string
 }
 
-// Render turns the desired state into iptables-restore input for the nat and mangle tables.
-// podNet is the weave allocation range (pods and bridges); bridge is the weave bridge name.
+// Render turns the desired state (load-balancer forwards and egress) into iptables-restore input for the nat,
+// mangle and filter tables. podNet is the weave allocation range (pods and bridges).
 // Each payload declares every chain it owns, so a restore with --noflush replaces exactly those.
-func Render(d Desired, podNet *net.IPNet, stale []string) Rules {
+func Render(d Desired, e EgressDesired, self string, podNet *net.IPNet, stale []string) Rules {
 	var nat bytes.Buffer
 	nat.WriteString("*nat\n")
-	fmt.Fprintf(&nat, ":%s - [0:0]\n", ForwardChain)
-	fmt.Fprintf(&nat, ":%s - [0:0]\n", PostChain)
+	for _, c := range []string{ForwardChain, PostChain, ClassChain} {
+		fmt.Fprintf(&nat, ":%s - [0:0]\n", c)
+	}
 	var body bytes.Buffer
 	seen := map[string]bool{}
 	for _, f := range d.Forwards {
@@ -93,12 +94,12 @@ func Render(d Desired, podNet *net.IPNet, stale []string) Rules {
 		// chains of ports no longer forwarded: emptied here, deleted after the restore
 		fmt.Fprintf(&nat, ":%s - [0:0]\n", c)
 	}
-	nat.Write(body.Bytes())
-	nat.WriteString("COMMIT\n")
 
 	var mangle bytes.Buffer
 	mangle.WriteString("*mangle\n")
-	fmt.Fprintf(&mangle, ":%s - [0:0]\n", MarkChain)
+	for _, c := range []string{MarkChain, MSSChain} {
+		fmt.Fprintf(&mangle, ":%s - [0:0]\n", c)
+	}
 	if len(d.Holders) > 0 {
 		// replies of a holder's connection carry its mark: the policy rule routes them back to it
 		fmt.Fprintf(&mangle, "-A %s -m conntrack --ctdir REPLY -m connmark ! --mark 0x0/0x%x -j CONNMARK --restore-mark --nfmask 0x%x --ctmask 0x%x\n",
@@ -110,8 +111,19 @@ func Render(d Desired, podNet *net.IPNet, stale []string) Rules {
 				MarkChain, bridge, h.Node.WeaveMAC, podNet, podNet, "holder "+h.Node.Name, markValue(h.Mark), markMask)
 		}
 	}
+
+	var filter bytes.Buffer
+	filter.WriteString("*filter\n")
+	for _, c := range []string{FwdChain, InChain} {
+		fmt.Fprintf(&filter, ":%s - [0:0]\n", c)
+	}
+	renderEgress(e, self, podNet, &body, &mangle, &filter)
+
+	nat.Write(body.Bytes())
+	nat.WriteString("COMMIT\n")
 	mangle.WriteString("COMMIT\n")
-	return Rules{Nat: nat.String(), Mangle: mangle.String()}
+	filter.WriteString("COMMIT\n")
+	return Rules{Nat: nat.String(), Mangle: mangle.String(), Filter: filter.String()}
 }
 
 // Datapath applies Desired to this node.
@@ -133,11 +145,20 @@ func NewDatapath(ipt *iptables.IPTables) *Datapath {
 	}}
 }
 
-// Apply makes the node's iptables chains, policy rules and routes exactly what d says.
-func (dp *Datapath) Apply(d Desired, podNet *net.IPNet) error {
+// Apply makes the node's iptables chains, tunnel, policy rules and routes exactly what d and e say.
+func (dp *Datapath) Apply(d Desired, e EgressDesired, self string, local Local) error {
+	podNet := local.PodNet
 	if err := dp.hooks(); err != nil {
 		return err
 	}
+	// the tunnels must exist before the routes into them (iptables rules may name a device that does not exist yet)
+	var peers []Node
+	if e.Tunnelled(self) {
+		peers = e.Peers(self)
+	}
+	if err := ensureTunnels(peers, local); err != nil {
+		return err
+	}
 	existing, err := dp.ipt.ListChains("nat")
 	if err != nil {
 		return err
@@ -154,28 +175,39 @@ func (dp *Datapath) Apply(d Desired, podNet *net.IPNet) error {
 			stale = append(stale, c)
 		}
 	}
-	r := Render(d, podNet, stale)
-	if err := dp.restore(r.Nat); err != nil {
-		return err
-	}
-	if err := dp.restore(r.Mangle); err != nil {
-		return err
+	r := Render(d, e, self, podNet, stale)
+	// filter first: the guards are in place before the marks and SNATs that they guard
+	for _, payload := range []string{r.Filter, r.Nat, r.Mangle} {
+		if err := dp.restore(payload); err != nil {
+			return err
+		}
 	}
 	for _, c := range stale {
 		if err := dp.ipt.DeleteChain("nat", c); err != nil {
 			return fmt.Errorf("deleting stale chain %s: %v", c, err)
 		}
 	}
-	return dp.routes(d.Holders)
+	if err := dp.routes(d.Holders); err != nil {
+		return err
+	}
+	return egressNet(e, self)
 }
 
-// hooks makes sure PREROUTING and POSTROUTING jump to our chains first (kube-proxy's and weave's must not see
-// held addresses' connections first).
+// hooks makes sure the built-in chains jump to ours: first (kube-proxy's and weave's rules must not see these
+// connections first), except ClassChain, which must come after kube-proxy's KUBE-SERVICES (whose DNAT ends the
+// nat walk, so a connection to a Service is never classed as going outside).
 func (dp *Datapath) hooks() error {
-	for _, h := range []struct{ table, from, chain string }{
-		{"nat", "PREROUTING", ForwardChain},
-		{"nat", "POSTROUTING", PostChain},
-		{"mangle", "PREROUTING", MarkChain},
+	for _, h := range []struct {
+		table, from, chain string
+		last               bool
+	}{
+		{"nat", "PREROUTING", ForwardChain, false},
+		{"nat", "POSTROUTING", PostChain, false},
+		{"nat", "PREROUTING", ClassChain, true},
+		{"mangle", "PREROUTING", MarkChain, false},
+		{"mangle", "FORWARD", MSSChain, false},
+		{"filter", "FORWARD", FwdChain, false},
+		{"filter", "INPUT", InChain, false},
 	} {
 		ok, err := dp.ipt.ChainExists(h.table, h.chain)
 		if err != nil {
@@ -191,6 +223,27 @@ func (dp *Datapath) hooks() error {
 		if err != nil {
 			return err
 		}
+		if h.last {
+			ours, kube := -1, -1
+			for i, r := range rules {
+				if strings.HasSuffix(r, "-j "+h.chain) {
+					ours = i
+				}
+				if strings.HasSuffix(r, "-j KUBE-SERVICES") {
+					kube = i
+				}
+			}
+			if ours > kube {
+				continue
+			}
+			if err := dp.ipt.DeleteIfExists(h.table, h.from, rule...); err != nil {
+				return err
+			}
+			if err := dp.ipt.Append(h.table, h.from, rule...); err != nil {
+				return err
+			}
+			continue
+		}
 		// rules[0] is the policy line; our jump must be rules[1]
 		if len(rules) > 1 && strings.HasSuffix(rules[1], "-j "+h.chain) {
 			continue
@@ -237,16 +290,15 @@ func (dp *Datapath) routes(holders []Holder) error {
 			return fmt.Errorf("removing stale rule to table %d: %v", r.Table, err)
 		}
 	}
+	existing, err := routesByTable(TableBase+1, TableBase+MaxMarks)
+	if err != nil {
+		return err
+	}
 	for table := TableBase + 1; table <= TableBase+MaxMarks; table++ {
 		h, ok := want[table]
 		if !ok {
-			routes, err := netlink.RouteListFiltered(unix.AF_INET, &netlink.Route{Table: table}, netlink.RT_FILTER_TABLE)
-			if err != nil {
-				return err
-			}
-			for _, rt := range routes {
-				rt := rt
-				if err := netlink.RouteDel(&rt); err != nil {
+			for _, rt := range existing[table] {
+				if err := delRoute(rt); err != nil {
 					return fmt.Errorf("removing stale route in table %d: %v", table, err)
 				}
 			}
@@ -274,3 +326,11 @@ func (dp *Datapath) routes(holders []Holder) error {
 	}
 	return nil
 }
+
+// delRoute deletes a route as RouteList returned it: a default route lists without Dst, which RouteDel refuses.
+func delRoute(rt netlink.Route) error {
+	if rt.Dst == nil {
+		rt.Dst = &net.IPNet{IP: net.IPv4zero, Mask: net.CIDRMask(0, 32)}
+	}
+	return netlink.RouteDel(&rt)
+}

lb/egress.go

@@ -0,0 +1,340 @@
+package lb
+
+import (
+	"fmt"
+	"net"
+	"sort"
+	"strings"
+
+	coreapi "k8s.io/api/core/v1"
+)
+
+// Port blocks: every node gets its own source ports on an EgressService's address, so the holder can route a
+// reply back to the member's node by port alone (no state on the holder). The range lies above the default
+// ephemeral range (32768-60999), so it never meets the holder's own connections from the address.
+const (
+	BlockBase  = 61000
+	BlockSize  = 512
+	BlockCount = 8 // 61000-65095
+)
+
+// Admission is a pod the CNI plugin announced at ADD, before the API shows its IP.
+type Admission struct {
+	Namespace, Name string
+}
+
+// Member is one pod on this node that an EgressService selects, and where its outside traffic goes.
+type Member struct {
+	IP      net.IP
+	Pod     string // namespace/name
+	Service string // namespace/name of its EgressService (empty if it is in several)
+	// Unreachable: its outside traffic is dropped (no holder, no block, several EgressServices, ...)
+	Unreachable bool
+	Address     net.IP // it leaves from Address:[From, To]
+	From, To    int32
+	Holder      Node // the node that emits it; Holder.Name == self: emitted here
+}
+
+// BlockRoute is, on the holder, the replies to Address:[From, To] going back to Peer.
+type BlockRoute struct {
+	Address  net.IP
+	From, To int32
+	Peer     Node
+}
+
+// EgressStatusWrite is a status this node writes (the holder, or the first node by name to clear it).
+type EgressStatusWrite struct {
+	Namespace, Name string
+	Status          EgressServiceStatus
+}
+
+// EgressDesired is this node's part of every EgressService.
+type EgressDesired struct {
+	Members   []Member
+	Routes    []BlockRoute
+	Status    []EgressStatusWrite
+	NodeAddrs []net.IP // the cluster's node addresses: members' traffic to them is not "outside"
+	Problems  []string
+}
+
+// Tunnelled reports whether this node sends or receives anything over the egress tunnel.
+func (e EgressDesired) Tunnelled(self string) bool {
+	if len(e.Routes) > 0 {
+		return true
+	}
+	for _, m := range e.Members {
+		if !m.Unreachable && m.Holder.Name != self {
+			return true
+		}
+	}
+	return false
+}
+
+// Peers are the nodes this node exchanges tunnelled packets with (sorted by name, deduplicated).
+func (e EgressDesired) Peers(self string) []Node {
+	seen := map[string]Node{}
+	for _, r := range e.Routes {
+		seen[r.Peer.Name] = r.Peer
+	}
+	for _, m := range e.Members {
+		if !m.Unreachable && m.Holder.Name != self {
+			seen[m.Holder.Name] = m.Holder
+		}
+	}
+	out := make([]Node, 0, len(seen))
+	for _, n := range seen {
+		out = append(out, n)
+	}
+	sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name })
+	return out
+}
+
+// AllocateBlocks keeps every valid block a node already has and gives each node without one the lowest free
+// block, in name order. A node beyond BlockCount gets none (its members are dropped, and that is reported).
+func AllocateBlocks(current []PortBlock, nodeNames []string) []PortBlock {
+	present := map[string]bool{}
+	for _, n := range nodeNames {
+		present[n] = true
+	}
+	used := map[int]bool{}
+	have := map[string]PortBlock{}
+	for _, b := range current {
+		k := int(b.From-BlockBase) / BlockSize
+		valid := present[b.Node] && b.From >= BlockBase && int(b.From-BlockBase)%BlockSize == 0 &&
+			k < BlockCount && b.To == b.From+BlockSize-1
+		if !valid || used[k] {
+			continue
+		}
+		if _, dup := have[b.Node]; dup {
+			continue
+		}
+		used[k] = true
+		have[b.Node] = b
+	}
+	names := append([]string(nil), nodeNames...)
+	sort.Strings(names)
+	for _, n := range names {
+		if _, ok := have[n]; ok {
+			continue
+		}
+		for k := 0; k < BlockCount; k++ {
+			if !used[k] {
+				used[k] = true
+				from := int32(BlockBase + k*BlockSize)
+				have[n] = PortBlock{Node: n, From: from, To: from + BlockSize - 1}
+				break
+			}
+		}
+	}
+	out := make([]PortBlock, 0, len(have))
+	for _, b := range have {
+		out = append(out, b)
+	}
+	sort.Slice(out, func(i, j int) bool { return out[i].Node < out[j].Node })
+	return out
+}
+
+func sameStatus(a, b EgressServiceStatus) bool {
+	if a.Host != b.Host || a.Address != b.Address || len(a.Blocks) != len(b.Blocks) {
+		return false
+	}
+	for i := range a.Blocks {
+		if a.Blocks[i] != b.Blocks[i] {
+			return false
+		}
+	}
+	return true
+}
+
+// holderOf is S1's rule for a Service's address: the one node that holds spec.loadBalancerIP.
+func holderOf(s *coreapi.Service, nodes []Node) (*Node, net.IP) {
+	want := net.ParseIP(s.Spec.LoadBalancerIP).To4()
+	if want == nil {
+		return nil, nil
+	}
+	var hs []Node
+	for _, n := range nodes {
+		if n.Holds(want) {
+			hs = append(hs, n)
+		}
+	}
+	if len(hs) != 1 {
+		return nil, want
+	}
+	return &hs[0], want
+}
+
+// selects reports whether a Service's selector (non-empty) matches the labels.
+func selects(s *coreapi.Service, labels map[string]string) bool {
+	if len(s.Spec.Selector) == 0 {
+		return false
+	}
+	for k, v := range s.Spec.Selector {
+		if labels[k] != v {
+			return false
+		}
+	}
+	return true
+}
+
+// ComputeEgress decides this node's part of every EgressService: the statuses it writes, the members it hosts and
+// where their traffic goes, and (as a holder) where replies to each node's port block go.
+// pods are this node's pods; admitted maps an IP to the pod the CNI plugin gave it (newer than the API may show).
+func ComputeEgress(self string, nodes []Node, services []*coreapi.Service, egresses []*EgressService,
+	pods []*coreapi.Pod, admitted map[string]Admission) EgressDesired {
+	var d EgressDesired
+	sorted := append([]Node(nil), nodes...)
+	sort.Slice(sorted, func(i, j int) bool { return sorted[i].Name < sorted[j].Name })
+	byName := map[string]Node{}
+	var names []string
+	for _, n := range sorted {
+		byName[n.Name] = n
+		names = append(names, n.Name)
+		d.NodeAddrs = append(d.NodeAddrs, n.Addresses...)
+	}
+	svcs := map[string]*coreapi.Service{}
+	for _, s := range services {
+		svcs[s.Namespace+"/"+s.Name] = s
+	}
+	es := append([]*EgressService(nil), egresses...)
+	sort.Slice(es, func(i, j int) bool { return es[i].Namespace+"/"+es[i].Name < es[j].Namespace+"/"+es[j].Name })
+
+	// valid[key]: the EgressService is in force: its Service is served from an address one node holds, and the
+	// holder has published that (status) - what members and holder both act on
+	valid := map[string]bool{}
+	for _, e := range es {
+		key := e.Namespace + "/" + e.Name
+		var want EgressServiceStatus
+		var holder *Node
+		var addr net.IP
+		s := svcs[key]
+		switch {
+		case e.Spec.Emit != EmitHolder:
+			d.Problems = append(d.Problems, fmt.Sprintf("egress %s: spec.emit must be %q (the only mode so far), got %q", key, EmitHolder, e.Spec.Emit))
+		case e.Spec.SourceIPBy != "" && e.Spec.SourceIPBy != SourceIPByLoadBalancerIP:
+			d.Problems = append(d.Problems, fmt.Sprintf("egress %s: spec.sourceIPBy must be %q, got %q", key, SourceIPByLoadBalancerIP, e.Spec.SourceIPBy))
+		case s == nil:
+			d.Problems = append(d.Problems, fmt.Sprintf("egress %s: no Service %s", key, key))
+		case !eligible(s):
+			d.Problems = append(d.Problems, fmt.Sprintf("egress %s: Service %s is not a LoadBalancer of class %s", key, key, Class))
+		default:
+			holder, addr = holderOf(s, sorted)
+			if holder == nil {
+				d.Problems = append(d.Problems, fmt.Sprintf("egress %s: no single node holds %s; members' outside traffic is dropped", key, s.Spec.LoadBalancerIP))
+			}
+		}
+		writer := ""
+		if holder != nil {
+			want = EgressServiceStatus{Host: holder.Name, Address: addr.String(), Blocks: AllocateBlocks(e.Status.Blocks, names)}
+			writer = holder.Name
+			if len(want.Blocks) < len(names) {
+				d.Problems = append(d.Problems, fmt.Sprintf("egress %s: only %d port blocks for %d nodes; members on nodes without one are dropped", key, BlockCount, len(names)))
+			}
+		} else if len(names) > 0 {
+			writer = names[0]
+		}
+		if writer == self && !sameStatus(e.Status, want) {
+			d.Status = append(d.Status, EgressStatusWrite{Namespace: e.Namespace, Name: e.Name, Status: want})
+		}
+		// in force only once the status in the API says what holds now (members and holder read the same thing)
+		if holder != nil && e.Status.Host == holder.Name && e.Status.Address == addr.String() {
+			valid[key] = true
+			if holder.Name == self {
+				for _, b := range e.Status.Blocks {
+					if b.Node == self {
+						continue
+					}
+					peer, ok := byName[b.Node]
+					if !ok || peer.WeaveIP == nil {
+						d.Problems = append(d.Problems, fmt.Sprintf("egress %s: node %s has no weave bridge published; its members' replies cannot reach it", key, b.Node))
+						continue
+					}
+					d.Routes = append(d.Routes, BlockRoute{Address: addr, From: b.From, To: b.To, Peer: peer})
+				}
+			}
+		}
+	}
+
+	// this node's pods, one per IP: the CNI plugin's latest word wins, then live pods over terminating ones
+	type candidate struct {
+		pod *coreapi.Pod
+		ip  string
+	}
+	byIP := map[string]*coreapi.Pod{}
+	podByKey := map[string]*coreapi.Pod{}
+	for _, p := range pods {
+		podByKey[p.Namespace+"/"+p.Name] = p
+	}
+	var cands []candidate
+	for _, p := range pods {
+		if p.Spec.NodeName != self || p.Spec.HostNetwork || p.Status.Phase == coreapi.PodSucceeded || p.Status.Phase == coreapi.PodFailed {
+			continue
+		}
+		if ip := net.ParseIP(p.Status.PodIP).To4(); ip != nil {
+			cands = append(cands, candidate{p, ip.String()})
+		}
+	}
+	for ip, a := range admitted {
+		if p := podByKey[a.Namespace+"/"+a.Name]; p != nil && p.Spec.NodeName == self && !p.Spec.HostNetwork {
+			cands = append(cands, candidate{p, ip})
+		}
+	}
+	for _, c := range cands {
+		if a, ok := admitted[c.ip]; ok && (a.Namespace != c.pod.Namespace || a.Name != c.pod.Name) {
+			continue // the IP was given to another pod since
+		}
+		cur := byIP[c.ip]
+		if cur == nil || (cur.DeletionTimestamp != nil && c.pod.DeletionTimestamp == nil) ||
+			(cur.DeletionTimestamp == nil) == (c.pod.DeletionTimestamp == nil) && c.pod.CreationTimestamp.After(cur.CreationTimestamp.Time) {
+			byIP[c.ip] = c.pod
+		}
+	}
+	ips := make([]string, 0, len(byIP))
+	for ip := range byIP {
+		ips = append(ips, ip)
+	}
+	sort.Strings(ips)
+	for _, ip := range ips {
+		p := byIP[ip]
+		var in []*EgressService
+		for _, e := range es {
+			if e.Namespace != p.Namespace {
+				continue
+			}
+			if s := svcs[e.Namespace+"/"+e.Name]; s != nil && selects(s, p.Labels) {
+				in = append(in, e)
+			}
+		}
+		if len(in) == 0 {
+			continue
+		}
+		m := Member{IP: net.ParseIP(ip).To4(), Pod: p.Namespace + "/" + p.Name, Unreachable: true}
+		if len(in) > 1 {
+			keys := make([]string, len(in))
+			for i, e := range in {
+				keys[i] = e.Namespace + "/" + e.Name
+			}
+			d.Problems = append(d.Problems, fmt.Sprintf("pod %s is selected by several EgressServices (%s); its outside traffic is dropped", m.Pod, strings.Join(keys, ", ")))
+			d.Members = append(d.Members, m)
+			continue
+		}
+		e := in[0]
+		m.Service = e.Namespace + "/" + e.Name
+		if valid[m.Service] {
+			holder := byName[e.Status.Host]
+			for _, b := range e.Status.Blocks {
+				if b.Node == self && (holder.Name == self || holder.WeaveIP != nil) {
+					m.Unreachable = false
+					m.Address = net.ParseIP(e.Status.Address).To4()
+					m.From, m.To = b.From, b.To
+					m.Holder = holder
+				}
+			}
+			if m.Unreachable {
+				d.Problems = append(d.Problems, fmt.Sprintf("pod %s: egress %s has no port block for %s (or its holder's bridge is unknown); its outside traffic is dropped", m.Pod, m.Service, self))
+			}
+		}
+		d.Members = append(d.Members, m)
+	}
+	return d
+}

lb/egress_api.go

@@ -0,0 +1,144 @@
+package lb
+
+import (
+	"context"
+	"encoding/json"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/runtime/serializer"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/rest"
+)
+
+// EgressService (reweave.levk.dev/v1alpha1) makes the pods a LoadBalancer Service selects leave the cluster
+// from the Service's address. It is named after the Service (same namespace, same name). Its semantics follow
+// OVN-Kubernetes' EgressService (k8s.ovn.org/v1, sourceIPBy: LoadBalancerIP) where they agree; they differ in
+// that membership is every pod the Service's selector matches (by pod IP, never by readiness), the holder is
+// the node that holds the address (no election), and a member's traffic to the outside leaves as the address
+// or not at all.
+type EgressService struct {
+	metav1.TypeMeta   `json:",inline"`
+	metav1.ObjectMeta `json:"metadata,omitempty"`
+	Spec              EgressServiceSpec   `json:"spec"`
+	Status            EgressServiceStatus `json:"status,omitempty"`
+}
+
+// EgressServiceSpec is what the operator asks for.
+type EgressServiceSpec struct {
+	// SourceIPBy is LoadBalancerIP: members leave from the Service's load-balancer address (the only mode in v1).
+	SourceIPBy string `json:"sourceIPBy,omitempty"`
+	// Emit is where members' packets leave the cluster. Explicit, no default: Holder (tunnelled to the node
+	// holding the address, which emits them) is the only mode in v1.
+	Emit string `json:"emit"`
+}
+
+// EgressServiceStatus is written by the holder alone.
+type EgressServiceStatus struct {
+	// Host is the node that holds the address and emits members' packets.
+	Host string `json:"host,omitempty"`
+	// Address is the address members leave from.
+	Address string `json:"address,omitempty"`
+	// Blocks gives every node its own source ports on Address, so the holder routes replies back by port.
+	Blocks []PortBlock `json:"blocks,omitempty"`
+}
+
+// PortBlock is the source-port range [From, To] on the address that one node's members use.
+type PortBlock struct {
+	Node string `json:"node"`
+	From int32  `json:"from"`
+	To   int32  `json:"to"`
+}
+
+// EgressServiceList is a list of EgressServices.
+type EgressServiceList struct {
+	metav1.TypeMeta `json:",inline"`
+	metav1.ListMeta `json:"metadata,omitempty"`
+	Items           []EgressService `json:"items"`
+}
+
+const (
+	// Group and Version of the EgressService API.
+	Group   = "reweave.levk.dev"
+	Version = "v1alpha1"
+	// EgressServices is the resource name.
+	EgressServices = "egressservices"
+
+	SourceIPByLoadBalancerIP = "LoadBalancerIP"
+	EmitHolder               = "Holder"
+)
+
+// GroupVersion of the EgressService API.
+var GroupVersion = schema.GroupVersion{Group: Group, Version: Version}
+
+// DeepCopyObject implements runtime.Object.
+func (e *EgressService) DeepCopyObject() runtime.Object { return e.DeepCopy() }
+
+// DeepCopy copies the whole object.
+func (e *EgressService) DeepCopy() *EgressService {
+	if e == nil {
+		return nil
+	}
+	out := *e
+	e.ObjectMeta.DeepCopyInto(&out.ObjectMeta)
+	out.Status.Blocks = append([]PortBlock(nil), e.Status.Blocks...)
+	return &out
+}
+
+// DeepCopyObject implements runtime.Object.
+func (l *EgressServiceList) DeepCopyObject() runtime.Object {
+	if l == nil {
+		return nil
+	}
+	out := *l
+	l.ListMeta.DeepCopyInto(&out.ListMeta)
+	out.Items = make([]EgressService, len(l.Items))
+	for i := range l.Items {
+		out.Items[i] = *l.Items[i].DeepCopy()
+	}
+	return &out
+}
+
+func addKnownTypes(s *runtime.Scheme) error {
+	s.AddKnownTypes(GroupVersion, &EgressService{}, &EgressServiceList{})
+	metav1.AddToGroupVersion(s, GroupVersion)
+	return nil
+}
+
+// NewEgressClient is a REST client for the EgressService API (no generated clientset: two plain types).
+func NewEgressClient(base *rest.Config) (*rest.RESTClient, error) {
+	scheme := runtime.NewScheme()
+	if err := addKnownTypes(scheme); err != nil {
+		return nil, err
+	}
+	cfg := *base
+	cfg.GroupVersion = &GroupVersion
+	cfg.APIPath = "/apis"
+	cfg.ContentType = runtime.ContentTypeJSON
+	cfg.NegotiatedSerializer = serializer.NewCodecFactory(scheme).WithoutConversion()
+	if cfg.UserAgent == "" {
+		cfg.UserAgent = rest.DefaultKubernetesUserAgent()
+	}
+	return rest.RESTClientFor(&cfg)
+}
+
+// PatchEgressStatus merge-patches an EgressService's status (nil fields are removed).
+func PatchEgressStatus(ctx context.Context, c rest.Interface, namespace, name string, st EgressServiceStatus) error {
+	body := map[string]interface{}{"host": nil, "address": nil, "blocks": nil}
+	if st.Host != "" {
+		body["host"] = st.Host
+	}
+	if st.Address != "" {
+		body["address"] = st.Address
+	}
+	if len(st.Blocks) > 0 {
+		body["blocks"] = st.Blocks
+	}
+	patch, err := json.Marshal(map[string]interface{}{"status": body})
+	if err != nil {
+		return err
+	}
+	return c.Patch(types.MergePatchType).Namespace(namespace).Resource(EgressServices).Name(name).
+		SubResource("status").Body(patch).Do(ctx).Error()
+}

lb/egress_datapath.go

@@ -0,0 +1,530 @@
+package lb
+
+import (
+	"bytes"
+	"fmt"
+	"hash/fnv"
+	"net"
+	"os"
+	"sort"
+	"strings"
+
+	"github.com/vishvananda/netlink"
+	"golang.org/x/sys/unix"
+)
+
+const (
+	// ClassChain (nat, appended to PREROUTING after kube-proxy's KUBE-SERVICES) marks members' new connections to
+	// the outside. kube-proxy's DNAT ends the nat walk, so a connection to a Service never gets here.
+	ClassChain = "REWEAVE-LB-CLASS"
+	// FwdChain (filter, first in FORWARD) drops members' outside packets that would not leave as their address.
+	FwdChain = "REWEAVE-LB-FWD"
+	// InChain (filter, first in INPUT) accepts the egress tunnels only from the cluster's nodes.
+	InChain = "REWEAVE-LB-IN"
+	// MSSChain (mangle, first in FORWARD) clamps TCP MSS into the tunnels.
+	MSSChain = "REWEAVE-LB-MSS"
+
+	// TunnelPrefix names the egress tunnels: one ipip device per peer node, between the two nodes' weave bridges,
+	// so a holder knows which node a member's packet came from by the device it arrives on.
+	TunnelPrefix = "rweg-"
+	// tunnelGlob matches every egress tunnel in an iptables rule.
+	tunnelGlob = "rweg+"
+	// legacyTunnel is the single shared tunnel of earlier builds (removed when found).
+	legacyTunnel = "reweave-eg"
+
+	// EgressMark marks a member's packets to the outside (connmark on its connections, fwmark on each packet).
+	EgressMark = uint32(0x100000)
+	// peerFlag, with a peer's index in the top byte, tags (on the holder) a connection that arrived from that peer's
+	// tunnel: replies and related ICMP errors carry it back there.
+	peerFlag = uint32(0x200000)
+	peerMask = markMask | peerFlag
+
+	// HolderRulePriority: replies to a block go to its node; ahead of the local table, which moves to LocalRulePriority,
+	// since the address is usually local on the holder.
+	HolderRulePriority = 90
+	LocalRulePriority  = 100
+	// MemberRulePriority: a member's marked packets go to its holder (or nowhere).
+	MemberRulePriority = 9100
+	// EgressTableBase + peer index routes into that peer's tunnel; EgressTableBase itself is unreachable.
+	EgressTableBase = 53000
+
+	ipipOverhead = 20
+)
+
+// portProtocols carry ports: their replies are routed by port block alone (stateless), and members' SNAT keeps
+// within the node's block. Everything else is routed back by the tag the holder put on the connection.
+var portProtocols = []struct {
+	name string
+	num  int
+}{{"tcp", unix.IPPROTO_TCP}, {"udp", unix.IPPROTO_UDP}, {"sctp", unix.IPPROTO_SCTP}}
+
+// TunnelName is the egress tunnel towards a peer node (stable, within the 15 characters a device name may have).
+func TunnelName(peer string) string {
+	h := fnv.New32a()
+	h.Write([]byte(peer))
+	return fmt.Sprintf("%s%08x", TunnelPrefix, h.Sum32())
+}
+
+func peerTag(idx uint32) uint32 { return markValue(idx) | peerFlag }
+
+// renderEgress appends the egress rules of each table to the bodies of S1's payloads.
+func renderEgress(e EgressDesired, self string, podNet *net.IPNet, nat, mangle, filter *bytes.Buffer) {
+	members := e.Members
+	if len(members) > 0 {
+		// what is not "outside": the pod network, multicast, the cluster's nodes
+		fmt.Fprintf(nat, "-A %s -d %s -j RETURN\n", ClassChain, podNet)
+		fmt.Fprintf(nat, "-A %s -d 224.0.0.0/4 -j RETURN\n", ClassChain)
+		seen := map[string]bool{}
+		for _, a := range e.NodeAddrs {
+			if !seen[a.String()] {
+				seen[a.String()] = true
+				fmt.Fprintf(nat, "-A %s -d %s/32 -j RETURN\n", ClassChain, a)
+			}
+		}
+		for _, m := range members {
+			fmt.Fprintf(nat, "-A %s -s %s/32 -m comment --comment %q -j CONNMARK --set-xmark 0x%x/0x%x\n", ClassChain, m.IP, m.Pod, EgressMark, EgressMark)
+			fmt.Fprintf(nat, "-A %s -s %s/32 -j MARK --set-xmark 0x%x/0x%x\n", ClassChain, m.IP, EgressMark, EgressMark)
+		}
+		// later packets of a member's outside connection carry the mark too (the nat table saw only the first)
+		fmt.Fprintf(mangle, "-A %s -m conntrack --ctdir ORIGINAL -m connmark --mark 0x%x/0x%x -j MARK --set-xmark 0x%x/0x%x\n",
+			MarkChain, EgressMark, EgressMark, EgressMark, EgressMark)
+	}
+	for _, m := range members {
+		if m.Unreachable {
+			fmt.Fprintf(filter, "-A %s -s %s/32 -m mark --mark 0x%x/0x%x -m comment --comment %q -j DROP\n", FwdChain, m.IP, EgressMark, EgressMark, m.Pod)
+			continue
+		}
+		out := "-o " + TunnelName(m.Holder.Name)
+		if m.Holder.Name == self {
+			// emitted here, from the address: never into the pod network unSNATed
+			out = "! -o " + bridge
+			fmt.Fprintf(filter, "-A %s -s %s/32 -o %s -m mark --mark 0x%x/0x%x -j DROP\n", FwdChain, m.IP, bridge, EgressMark, EgressMark)
+		}
+		// within this node's block for what has ports; the address alone for the rest (the holder tells those
+		// apart by the tunnel they arrive on)
+		for _, p := range portProtocols {
+			fmt.Fprintf(nat, "-A %s -s %s/32 %s -m mark --mark 0x%x/0x%x -p %s -j SNAT --to-source %s:%d-%d\n",
+				PostChain, m.IP, out, EgressMark, EgressMark, p.name, m.Address, m.From, m.To)
+		}
+		// ICMP queries (ping) too: NAT treats the echo identifier as a port, and without a block of their own two
+		// nodes' pods (whose small, alike PIDs make alike identifiers) collide on the holder
+		fmt.Fprintf(nat, "-A %s -s %s/32 %s -m mark --mark 0x%x/0x%x -p icmp -j SNAT --to-source %s:%d-%d\n",
+			PostChain, m.IP, out, EgressMark, EgressMark, m.Address, m.From, m.To)
+		fmt.Fprintf(nat, "-A %s -s %s/32 %s -m mark --mark 0x%x/0x%x -j SNAT --to-source %s\n",
+			PostChain, m.IP, out, EgressMark, EgressMark, m.Address)
+		fmt.Fprintf(filter, "-A %s -s %s/32 %s -m mark --mark 0x%x/0x%x -j RETURN\n", FwdChain, m.IP, out, EgressMark, EgressMark)
+		// a member's outside traffic leaves no other way
+		fmt.Fprintf(filter, "-A %s -s %s/32 -m mark --mark 0x%x/0x%x -m comment --comment %q -j DROP\n", FwdChain, m.IP, EgressMark, EgressMark, m.Pod)
+	}
+	if !e.Tunnelled(self) {
+		return
+	}
+	tags := peerIndex(e, self)
+
+	// as holder: tag each connection arriving from a peer's tunnel with that peer, once; replies and related ICMP
+	// errors restore the tag, and the policy rule for it sends them back into that tunnel
+	byPeer := map[string][]BlockRoute{}
+	var peerNames []string
+	for _, r := range e.Routes {
+		if _, ok := byPeer[r.Peer.Name]; !ok {
+			peerNames = append(peerNames, r.Peer.Name)
+		}
+		byPeer[r.Peer.Name] = append(byPeer[r.Peer.Name], r)
+	}
+	sort.Strings(peerNames)
+	if len(peerNames) > 0 {
+		fmt.Fprintf(mangle, "-A %s -m conntrack --ctdir REPLY -m connmark --mark 0x%x/0x%x -j CONNMARK --restore-mark --nfmask 0x%x --ctmask 0x%x\n",
+			MarkChain, peerFlag, peerFlag, peerMask, peerMask)
+	}
+	for _, name := range peerNames {
+		dev, tag := TunnelName(name), peerTag(tags[name])
+		fmt.Fprintf(mangle, "-A %s -i %s -m conntrack --ctstate NEW -m connmark --mark 0x0/0x%x -m comment --comment %q -j CONNMARK --set-xmark 0x%x/0x%x\n",
+			MarkChain, dev, peerMask, "from "+name, tag, peerMask)
+		// a connection another node already holds (a port-less protocol to the same remote): never cross-delivered
+		fmt.Fprintf(filter, "-A %s -i %s -m conntrack --ctdir ORIGINAL -m connmark ! --mark 0x%x/0x%x -j DROP\n", FwdChain, dev, tag, peerMask)
+		addrs := map[string]bool{}
+		for _, r := range byPeer[name] {
+			for _, p := range portProtocols {
+				fmt.Fprintf(filter, "-A %s -i %s -m conntrack --ctdir ORIGINAL -s %s/32 -p %s -m %s --sport %d:%d -m comment --comment %q -j RETURN\n",
+					FwdChain, dev, r.Address, p.name, p.name, r.From, r.To, "block of "+name)
+			}
+			addrs[r.Address.String()] = true
+		}
+		for _, a := range sortedSet(addrs) {
+			// outside its own block a node may not use the ports of others
+			for _, p := range portProtocols {
+				fmt.Fprintf(filter, "-A %s -i %s -m conntrack --ctdir ORIGINAL -s %s/32 -p %s -j DROP\n", FwdChain, dev, a, p.name)
+			}
+			fmt.Fprintf(filter, "-A %s -i %s -m conntrack --ctdir ORIGINAL -s %s/32 -j RETURN\n", FwdChain, dev, a)
+		}
+	}
+	// as a member's node: replies (and related errors) from the holder; nothing else comes back
+	addrs := map[string]bool{}
+	for _, m := range members {
+		if !m.Unreachable && m.Holder.Name != self {
+			addrs[m.Address.String()] = true
+		}
+	}
+	for _, a := range sortedSet(addrs) {
+		// still addressed to the address after nat: no member connection matched it
+		fmt.Fprintf(filter, "-A %s -i %s -d %s/32 -j DROP\n", FwdChain, tunnelGlob, a)
+	}
+	fmt.Fprintf(filter, "-A %s -i %s -m conntrack --ctstate ESTABLISHED,RELATED -j RETURN\n", FwdChain, tunnelGlob)
+	fmt.Fprintf(filter, "-A %s -i %s -j DROP\n", FwdChain, tunnelGlob)
+	// the tunnels themselves only from the cluster's nodes' bridges
+	for _, p := range e.Peers(self) {
+		fmt.Fprintf(filter, "-A %s -p 4 -s %s/32 -m mac --mac-source %s -j RETURN\n", InChain, p.WeaveIP, p.WeaveMAC)
+	}
+	fmt.Fprintf(filter, "-A %s -p 4 -j DROP\n", InChain)
+	fmt.Fprintf(mangle, "-A %s -o %s -p tcp -m tcp --tcp-flags SYN,RST SYN -j TCPMSS --clamp-mss-to-pmtu\n", MSSChain, tunnelGlob)
+}
+
+func sortedSet(m map[string]bool) []string {
+	out := make([]string, 0, len(m))
+	for k := range m {
+		out = append(out, k)
+	}
+	sort.Strings(out)
+	return out
+}
+
+// peerIndex gives each tunnel peer a stable index (1..MaxMarks): its tag and its routing table.
+func peerIndex(e EgressDesired, self string) map[string]uint32 {
+	out := map[string]uint32{}
+	for _, h := range assignMarks(e.Peers(self)) {
+		out[h.Node.Name] = h.Mark
+	}
+	return out
+}
+
+// peerTables gives each tunnel peer its routing table (EgressTableBase + its index).
+func peerTables(e EgressDesired, self string) map[string]int {
+	out := map[string]int{}
+	for name, idx := range peerIndex(e, self) {
+		out[name] = EgressTableBase + int(idx)
+	}
+	return out
+}
+
+// egressRule is a policy rule this package owns, in a comparable form.
+type egressRule struct {
+	priority   int
+	src, dst   string
+	ipproto    int
+	from, to   uint16
+	mark, mask uint32
+	table      int
+}
+
+func (r egressRule) key() string {
+	return fmt.Sprintf("%d|%s|%s|%d|%d-%d|%x/%x|%d", r.priority, r.src, r.dst, r.ipproto, r.from, r.to, r.mark, r.mask, r.table)
+}
+
+// wantRules is every egress policy rule this node needs.
+func wantRules(e EgressDesired, self string) []egressRule {
+	tables := peerTables(e, self)
+	tags := peerIndex(e, self)
+	var out []egressRule
+	tagged := map[string]bool{}
+	for _, r := range e.Routes {
+		for _, p := range portProtocols {
+			out = append(out, egressRule{priority: HolderRulePriority, dst: r.Address.String() + "/32", ipproto: p.num,
+				from: uint16(r.From), to: uint16(r.To), table: tables[r.Peer.Name]})
+		}
+		if !tagged[r.Peer.Name] {
+			// whatever the tag says came from that peer (replies without ports, ICMP errors) goes back to it
+			tagged[r.Peer.Name] = true
+			out = append(out, egressRule{priority: HolderRulePriority, mark: peerTag(tags[r.Peer.Name]), mask: peerMask, table: tables[r.Peer.Name]})
+		}
+	}
+	for _, m := range e.Members {
+		if !m.Unreachable && m.Holder.Name == self {
+			continue // emitted here: the main table already sends it out
+		}
+		table := EgressTableBase
+		if !m.Unreachable {
+			table = tables[m.Holder.Name]
+		}
+		out = append(out, egressRule{priority: MemberRulePriority, src: m.IP.String() + "/32", mark: EgressMark, mask: EgressMark, table: table})
+	}
+	sort.Slice(out, func(i, j int) bool { return out[i].key() < out[j].key() })
+	return out
+}
+
+func ruleOf(r netlink.Rule) egressRule {
+	er := egressRule{priority: r.Priority, ipproto: r.IPProto, table: r.Table}
+	if r.Src != nil {
+		er.src = r.Src.String()
+	}
+	if r.Dst != nil {
+		er.dst = r.Dst.String()
+	}
+	if r.Dport != nil {
+		er.from, er.to = r.Dport.Start, r.Dport.End
+	}
+	if r.Mark > 0 {
+		er.mark = uint32(r.Mark)
+		er.mask = uint32(r.Mask)
+	}
+	if er.ipproto < 0 {
+		er.ipproto = 0
+	}
+	return er
+}
+
+// routesByTable lists, in one dump, the IPv4 routes of tables lo..hi.
+func routesByTable(lo, hi int) (map[int][]netlink.Route, error) {
+	all, err := netlink.RouteListFiltered(unix.AF_INET, &netlink.Route{Table: unix.RT_TABLE_UNSPEC}, netlink.RT_FILTER_TABLE)
+	if err != nil {
+		return nil, err
+	}
+	out := map[int][]netlink.Route{}
+	for _, rt := range all {
+		if rt.Table >= lo && rt.Table <= hi {
+			out[rt.Table] = append(out[rt.Table], rt)
+		}
+	}
+	return out, nil
+}
+
+// egressNet makes the local rule's place, the policy rules and the tunnel tables exactly what e needs (the
+// tunnels themselves are made first, by Apply).
+func egressNet(e EgressDesired, self string) error {
+	if len(e.Routes) > 0 {
+		if err := moveLocalRule(); err != nil {
+			return err
+		}
+	}
+	tables := peerTables(e, self)
+	peerOf := map[int]Node{}
+	for _, p := range e.Peers(self) {
+		peerOf[tables[p.Name]] = p
+	}
+	unreachable := false
+	for _, m := range e.Members {
+		unreachable = unreachable || m.Unreachable
+	}
+	existing, err := routesByTable(EgressTableBase, EgressTableBase+MaxMarks)
+	if err != nil {
+		return err
+	}
+	for table := EgressTableBase; table <= EgressTableBase+MaxMarks; table++ {
+		var want *netlink.Route
+		if p, ok := peerOf[table]; ok {
+			link, err := netlink.LinkByName(TunnelName(p.Name))
+			if err != nil {
+				return fmt.Errorf("egress tunnel to %s: %v", p.Name, err)
+			}
+			want = &netlink.Route{LinkIndex: link.Attrs().Index, Dst: &net.IPNet{IP: net.IPv4zero, Mask: net.CIDRMask(0, 32)},
+				Scope: netlink.SCOPE_LINK, Table: table}
+		} else if table == EgressTableBase && unreachable {
+			want = &netlink.Route{Dst: &net.IPNet{IP: net.IPv4zero, Mask: net.CIDRMask(0, 32)}, Table: table, Type: unix.RTN_UNREACHABLE}
+		}
+		keep := false
+		for _, rt := range existing[table] {
+			if want != nil && rt.Type == want.Type &&
+				(want.Type == unix.RTN_UNREACHABLE || rt.LinkIndex == want.LinkIndex && rt.Gw == nil) {
+				keep = true
+				continue
+			}
+			if err := delRoute(rt); err != nil {
+				return fmt.Errorf("removing stale route in table %d: %v", table, err)
+			}
+		}
+		if want != nil && !keep {
+			if err := netlink.RouteReplace(want); err != nil {
+				return fmt.Errorf("route in table %d: %v", table, err)
+			}
+		}
+	}
+	want := map[string]egressRule{}
+	for _, r := range wantRules(e, self) {
+		want[r.key()] = r
+	}
+	rules, err := netlink.RuleList(unix.AF_INET)
+	if err != nil {
+		return err
+	}
+	have := map[string]bool{}
+	for _, r := range rules {
+		if r.Priority != HolderRulePriority && r.Priority != MemberRulePriority {
+			continue
+		}
+		k := ruleOf(r).key()
+		if _, ok := want[k]; ok && !have[k] {
+			have[k] = true
+			continue
+		}
+		r := r
+		if err := netlink.RuleDel(&r); err != nil {
+			return fmt.Errorf("removing stale egress rule: %v", err)
+		}
+	}
+	for k, r := range want {
+		if have[k] {
+			continue
+		}
+		rule := netlink.NewRule()
+		rule.Priority = r.priority
+		rule.Table = r.table
+		if r.src != "" {
+			_, rule.Src, _ = net.ParseCIDR(r.src)
+		}
+		if r.dst != "" {
+			_, rule.Dst, _ = net.ParseCIDR(r.dst)
+		}
+		if r.ipproto != 0 {
+			rule.IPProto = r.ipproto
+			rule.Dport = netlink.NewRulePortRange(r.from, r.to)
+		}
+		if r.mark != 0 {
+			rule.Mark = int(r.mark)
+			rule.Mask = int(r.mask)
+		}
+		if err := netlink.RuleAdd(rule); err != nil {
+			return fmt.Errorf("adding egress rule %s: %v", k, err)
+		}
+	}
+	return nil
+}
+
+// moveLocalRule puts the local table's lookup after HolderRulePriority (it is at 0 by default), so the holder can
+// route replies to its own address onwards to their node instead of delivering them to itself.
+func moveLocalRule() error {
+	rules, err := netlink.RuleList(unix.AF_INET)
+	if err != nil {
+		return err
+	}
+	var at0, atLocal bool
+	for _, r := range rules {
+		if r.Table == unix.RT_TABLE_LOCAL && r.Src == nil && r.Dst == nil && r.Mark <= 0 {
+			switch {
+			case r.Priority <= 0:
+				// the kernel leaves out the priority attribute when it is 0: listed as unset (-1)
+				at0 = true
+			case r.Priority == LocalRulePriority:
+				atLocal = true
+			}
+		}
+	}
+	if !atLocal {
+		rule := netlink.NewRule()
+		rule.Priority = LocalRulePriority
+		rule.Table = unix.RT_TABLE_LOCAL
+		if err := netlink.RuleAdd(rule); err != nil {
+			return fmt.Errorf("adding local table rule at %d: %v", LocalRulePriority, err)
+		}
+	}
+	if at0 {
+		rule := netlink.NewRule()
+		rule.Priority = 0
+		rule.Table = unix.RT_TABLE_LOCAL
+		if err := netlink.RuleDel(rule); err != nil {
+			return fmt.Errorf("removing local table rule at 0: %v", err)
+		}
+	}
+	return nil
+}
+
+// ensureTunnels makes one ipip tunnel per peer (this node's bridge address to the peer's), and removes every
+// other egress tunnel (including the shared one of earlier builds).
+func ensureTunnels(peers []Node, local Local) error {
+	links, err := netlink.LinkList()
+	if err != nil {
+		return err
+	}
+	want := map[string]Node{}
+	for _, p := range peers {
+		want[TunnelName(p.Name)] = p
+	}
+	have := map[string]netlink.Link{}
+	for _, l := range links {
+		name := l.Attrs().Name
+		if name == legacyTunnel || strings.HasPrefix(name, TunnelPrefix) {
+			if _, ok := want[name]; ok {
+				have[name] = l
+				continue
+			}
+			if err := netlink.LinkDel(l); err != nil {
+				return fmt.Errorf("removing egress tunnel %s: %v", name, err)
+			}
+		}
+	}
+	if len(want) == 0 {
+		return nil
+	}
+	if local.WeaveIP == nil {
+		return fmt.Errorf("egress tunnels need the weave bridge's address")
+	}
+	br, err := netlink.LinkByName(bridge)
+	if err != nil {
+		return fmt.Errorf("weave bridge: %v", err)
+	}
+	mtu := br.Attrs().MTU - ipipOverhead
+	names := make([]string, 0, len(want))
+	for n := range want {
+		names = append(names, n)
+	}
+	sort.Strings(names)
+	for _, name := range names {
+		peer := want[name]
+		link := have[name]
+		if link != nil {
+			t, ok := link.(*netlink.Iptun)
+			if !ok || !t.Local.Equal(local.WeaveIP) || !t.Remote.Equal(peer.WeaveIP) || t.Attrs().MTU != mtu {
+				if err := netlink.LinkDel(link); err != nil {
+					return err
+				}
+				link = nil
+			}
+		}
+		if link == nil {
+			t := &netlink.Iptun{LinkAttrs: netlink.LinkAttrs{Name: name, MTU: mtu}, Local: local.WeaveIP, Remote: peer.WeaveIP, PMtuDisc: 1}
+			if err := netlink.LinkAdd(t); err != nil {
+				return fmt.Errorf("creating egress tunnel %s to %s: %v", name, peer.Name, err)
+			}
+			if link, err = netlink.LinkByName(name); err != nil {
+				return err
+			}
+		}
+		if err := tunnelSetup(link, local); err != nil {
+			return fmt.Errorf("%s: %v", name, err)
+		}
+	}
+	return nil
+}
+
+// tunnelSetup: the holder receives members' packets from its own address (accept_local); reverse-path filtering
+// rejects what arrives on a device with no IPv4 address unless the reverse route leaves by that very device, so
+// the tunnel carries this node's bridge address (never on the wire from here: the tunnel sets the outer source,
+// and what crosses it is forwarded).
+func tunnelSetup(link netlink.Link, local Local) error {
+	name := link.Attrs().Name
+	for k, v := range map[string]string{"accept_local": "1", "rp_filter": "2"} {
+		if err := os.WriteFile("/proc/sys/net/ipv4/conf/"+name+"/"+k, []byte(v), 0o644); err != nil {
+			return fmt.Errorf("sysctl %s: %v", k, err)
+		}
+	}
+	addrs, err := netlink.AddrList(link, unix.AF_INET)
+	if err != nil {
+		return err
+	}
+	has := false
+	for _, a := range addrs {
+		if a.IP.Equal(local.WeaveIP) {
+			has = true
+			continue
+		}
+		a := a
+		if err := netlink.AddrDel(link, &a); err != nil {
+			return fmt.Errorf("removing stale address %s: %v", a.IPNet, err)
+		}
+	}
+	if !has {
+		addr := &netlink.Addr{IPNet: &net.IPNet{IP: local.WeaveIP, Mask: net.CIDRMask(32, 32)}, Scope: unix.RT_SCOPE_LINK}
+		if err := netlink.AddrAdd(link, addr); err != nil {
+			return fmt.Errorf("address: %v", err)
+		}
+	}
+	return netlink.LinkSetUp(link)
+}

lb/egress_test.go

@@ -0,0 +1,355 @@
+package lb
+
+import (
+	"fmt"
+	"net"
+	"strings"
+	"testing"
+	"time"
+
+	coreapi "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+// the S1 fixtures: cp, w1 (holds 86.107.168.15), w2
+
+func egressFor(name string, status EgressServiceStatus) *EgressService {
+	return &EgressService{ObjectMeta: metav1.ObjectMeta{Namespace: "gitea", Name: name},
+		Spec: EgressServiceSpec{Emit: EmitHolder}, Status: status}
+}
+
+func mailSvc() *coreapi.Service {
+	return svc("mail", "86.107.168.15", func(s *coreapi.Service) {
+		s.Spec.Selector = map[string]string{"app": "mail"}
+	})
+}
+
+func pod(name, node, ip string, labels map[string]string, mutate ...func(*coreapi.Pod)) *coreapi.Pod {
+	p := &coreapi.Pod{ObjectMeta: metav1.ObjectMeta{Namespace: "gitea", Name: name, Labels: labels,
+		CreationTimestamp: metav1.NewTime(time.Unix(1000, 0))},
+		Spec: coreapi.PodSpec{NodeName: node}, Status: coreapi.PodStatus{PodIP: ip, Phase: coreapi.PodRunning}}
+	for _, m := range mutate {
+		m(p)
+	}
+	return p
+}
+
+var mailLabels = map[string]string{"app": "mail"}
+
+// the status the holder (w1) writes for three nodes, blocks in name order
+func written() EgressServiceStatus {
+	return EgressServiceStatus{Host: "w1", Address: "86.107.168.15", Blocks: []PortBlock{
+		{Node: "cp", From: 61000, To: 61511}, {Node: "w1", From: 61512, To: 62023}, {Node: "w2", From: 62024, To: 62535}}}
+}
+
+func TestBlocksAreStableAndCapped(t *testing.T) {
+	got := AllocateBlocks(nil, []string{"w2", "cp", "w1"})
+	if !sameStatus(EgressServiceStatus{Blocks: got}, EgressServiceStatus{Blocks: written().Blocks}) {
+		t.Fatalf("fresh blocks: %+v", got)
+	}
+	// a node leaves, another joins: survivors keep their blocks, the newcomer takes the freed lowest one
+	got = AllocateBlocks(got, []string{"w1", "w2", "w3"})
+	want := []PortBlock{{"w1", 61512, 62023}, {"w2", 62024, 62535}, {"w3", 61000, 61511}}
+	if !sameStatus(EgressServiceStatus{Blocks: got}, EgressServiceStatus{Blocks: want}) {
+		t.Fatalf("after churn: %+v", got)
+	}
+	// misaligned or duplicate blocks are not kept
+	got = AllocateBlocks([]PortBlock{{"w1", 61001, 61512}, {"w2", 61000, 61511}, {"w3", 61000, 61511}}, []string{"w1", "w2", "w3"})
+	for _, b := range got {
+		if (b.From-BlockBase)%BlockSize != 0 || b.To != b.From+BlockSize-1 {
+			t.Fatalf("invalid block kept: %+v", got)
+		}
+	}
+	if got[1].Node != "w2" || got[1].From != 61000 {
+		t.Fatalf("first valid claim wins: %+v", got)
+	}
+	var many []string
+	for i := 0; i < BlockCount+2; i++ {
+		many = append(many, string(rune('a'+i)))
+	}
+	if n := len(AllocateBlocks(nil, many)); n != BlockCount {
+		t.Fatalf("more blocks than exist: %d", n)
+	}
+	for _, b := range AllocateBlocks(nil, many) {
+		if b.To > 65535 || b.From < 61000 {
+			t.Fatalf("block outside the range: %+v", b)
+		}
+	}
+}
+
+func TestHolderWritesStatusAndMembersWaitForIt(t *testing.T) {
+	e := egressFor("mail", EgressServiceStatus{})
+	pods := []*coreapi.Pod{pod("mail-0", "w2", "10.44.0.9", mailLabels)}
+	d := ComputeEgress("w1", nodes, []*coreapi.Service{mailSvc()}, []*EgressService{e}, nil, nil)
+	if len(d.Status) != 1 || !sameStatus(d.Status[0].Status, written()) {
+		t.Fatalf("holder's status: %+v", d.Status)
+	}
+	// until the status is in the API, a member's outside traffic goes nowhere (fail closed)
+	d = ComputeEgress("w2", nodes, []*coreapi.Service{mailSvc()}, []*EgressService{e}, pods, nil)
+	if len(d.Status) != 0 || len(d.Members) != 1 || !d.Members[0].Unreachable {
+		t.Fatalf("member before status: %+v", d)
+	}
+	if d.Tunnelled("w2") {
+		t.Fatalf("no tunnel for an unreachable member")
+	}
+}
+
+func TestMemberTunnelsToHolderWhichRoutesItsBlockBack(t *testing.T) {
+	e := egressFor("mail", written())
+	svcs := []*coreapi.Service{mailSvc()}
+	pods := []*coreapi.Pod{
+		pod("mail-0", "w2", "10.44.0.9", mailLabels),
+		pod("web", "w2", "10.44.0.10", map[string]string{"app": "web"}), // not a member (A4)
+	}
+	d := ComputeEgress("w2", nodes, svcs, []*EgressService{e}, pods, nil)
+	if len(d.Members) != 1 || d.Members[0].Pod != "gitea/mail-0" {
+		t.Fatalf("members: %+v", d.Members)
+	}
+	m := d.Members[0]
+	if m.Unreachable || !m.Address.Equal(net.ParseIP("86.107.168.15")) || m.From != 62024 || m.To != 62535 || m.Holder.Name != "w1" {
+		t.Fatalf("member path: %+v", m)
+	}
+	if !d.Tunnelled("w2") || len(d.Peers("w2")) != 1 || d.Peers("w2")[0].Name != "w1" {
+		t.Fatalf("w2 tunnels to w1 only: %+v", d.Peers("w2"))
+	}
+	if len(d.Status) != 0 || len(d.Routes) != 0 {
+		t.Fatalf("only the holder writes and routes: %+v", d)
+	}
+	// the holder: blocks of the other nodes route back to them; its own block is plain conntrack
+	h := ComputeEgress("w1", nodes, svcs, []*EgressService{e}, nil, nil)
+	if len(h.Status) != 0 {
+		t.Fatalf("status already right: %+v", h.Status)
+	}
+	if len(h.Routes) != 2 || h.Routes[0].Peer.Name != "cp" || h.Routes[1].Peer.Name != "w2" || h.Routes[1].From != 62024 {
+		t.Fatalf("holder routes: %+v", h.Routes)
+	}
+	// a member on the holder leaves from there
+	hm := ComputeEgress("w1", nodes, svcs, []*EgressService{e}, []*coreapi.Pod{pod("mail-1", "w1", "10.40.0.4", mailLabels)}, nil)
+	if len(hm.Members) != 1 || hm.Members[0].Holder.Name != "w1" || hm.Members[0].From != 61512 {
+		t.Fatalf("member on the holder: %+v", hm.Members)
+	}
+}
+
+func TestEgressFailsClosed(t *testing.T) {
+	pods := []*coreapi.Pod{pod("mail-0", "w2", "10.44.0.9", mailLabels)}
+	// no node holds the address: status cleared (by the first node by name), member dropped
+	unheld := svc("mail", "86.107.168.99", func(s *coreapi.Service) { s.Spec.Selector = mailLabels })
+	for _, self := range []string{"cp", "w2"} {
+		d := ComputeEgress(self, nodes, []*coreapi.Service{unheld}, []*EgressService{egressFor("mail", written())}, pods, nil)
+		clears := len(d.Status) == 1 && d.Status[0].Status.Host == ""
+		if clears != (self == "cp") {
+			t.Fatalf("%s clears=%v: %+v", self, clears, d.Status)
+		}
+		if self == "w2" && (len(d.Members) != 1 || !d.Members[0].Unreachable) {
+			t.Fatalf("member of an unheld address: %+v", d.Members)
+		}
+	}
+	// two EgressServices select the pod: dropped, reported
+	other := svc("mail2", "86.107.168.15", func(s *coreapi.Service) { s.Spec.Selector = mailLabels })
+	d := ComputeEgress("w2", nodes, []*coreapi.Service{mailSvc(), other},
+		[]*EgressService{egressFor("mail", written()), egressFor("mail2", written())}, pods, nil)
+	if len(d.Members) != 1 || !d.Members[0].Unreachable || !strings.Contains(strings.Join(d.Problems, "\n"), "several EgressServices") {
+		t.Fatalf("double member: %+v", d)
+	}
+	// a spec this agent does not implement is refused, not guessed at
+	bad := egressFor("mail", written())
+	bad.Spec.Emit = "Direct"
+	d = ComputeEgress("w2", nodes, []*coreapi.Service{mailSvc()}, []*EgressService{bad}, pods, nil)
+	if len(d.Members) != 1 || !d.Members[0].Unreachable {
+		t.Fatalf("unknown emit mode: %+v", d)
+	}
+	// hostNetwork pods are never members (their traffic is the node's)
+	hn := pod("mail-h", "w2", "86.107.168.88", mailLabels, func(p *coreapi.Pod) { p.Spec.HostNetwork = true })
+	d = ComputeEgress("w2", nodes, []*coreapi.Service{mailSvc()}, []*EgressService{egressFor("mail", written())}, []*coreapi.Pod{hn}, nil)
+	if len(d.Members) != 0 {
+		t.Fatalf("hostNetwork pod became a member: %+v", d.Members)
+	}
+}
+
+func TestAdmissionDecidesWhoOwnsAnIP(t *testing.T) {
+	e := egressFor("mail", written())
+	svcs := []*coreapi.Service{mailSvc()}
+	old := pod("mail-0", "w2", "10.44.0.9", mailLabels, func(p *coreapi.Pod) {
+		now := metav1.Now()
+		p.DeletionTimestamp = &now
+	})
+	newer := pod("web", "w2", "", map[string]string{"app": "web"}) // API has no IP for it yet
+	// the CNI plugin just gave 10.44.0.9 to web: the terminating member must not keep it (A4)
+	d := ComputeEgress("w2", nodes, svcs, []*EgressService{e}, []*coreapi.Pod{old, newer},
+		map[string]Admission{"10.44.0.9": {Namespace: "gitea", Name: "web"}})
+	if len(d.Members) != 0 {
+		t.Fatalf("reused IP kept the old member's rules: %+v", d.Members)
+	}
+	// and the other way round: a member admitted before the API shows its IP is protected at once (A1)
+	fresh := pod("mail-1", "w2", "", mailLabels)
+	d = ComputeEgress("w2", nodes, svcs, []*EgressService{e}, []*coreapi.Pod{fresh},
+		map[string]Admission{"10.44.0.11": {Namespace: "gitea", Name: "mail-1"}})
+	if len(d.Members) != 1 || d.Members[0].IP.String() != "10.44.0.11" || d.Members[0].Unreachable {
+		t.Fatalf("admitted member: %+v", d.Members)
+	}
+	// without an admission, a live pod wins a shared IP over a terminating one
+	live := pod("web", "w2", "10.44.0.9", map[string]string{"app": "web"})
+	d = ComputeEgress("w2", nodes, svcs, []*EgressService{e}, []*coreapi.Pod{old, live}, nil)
+	if len(d.Members) != 0 {
+		t.Fatalf("terminating member won the IP: %+v", d.Members)
+	}
+}
+
+func TestRenderedEgressRules(t *testing.T) {
+	_, podNet, _ := net.ParseCIDR("10.32.0.0/12")
+	e := egressFor("mail", written())
+	svcs := []*coreapi.Service{mailSvc()}
+	toW1, toW2, toCP := TunnelName("w1"), TunnelName("w2"), TunnelName("cp")
+	member := ComputeEgress("w2", nodes, svcs, []*EgressService{e}, []*coreapi.Pod{pod("mail-0", "w2", "10.44.0.9", mailLabels)}, nil)
+	r := Render(Desired{}, member, "w2", podNet, nil)
+	for _, want := range []string{
+		"-A REWEAVE-LB-CLASS -d 10.32.0.0/12 -j RETURN\n",
+		"-A REWEAVE-LB-CLASS -d 86.107.168.24/32 -j RETURN\n", // a node is not outside
+		"-A REWEAVE-LB-CLASS -s 10.44.0.9/32 -m comment --comment \"gitea/mail-0\" -j CONNMARK --set-xmark 0x100000/0x100000\n",
+		"-A REWEAVE-LB-CLASS -s 10.44.0.9/32 -j MARK --set-xmark 0x100000/0x100000\n",
+		// port protocols within this node's block, into the holder's own tunnel
+		"-A REWEAVE-LB-POST -s 10.44.0.9/32 -o " + toW1 + " -m mark --mark 0x100000/0x100000 -p tcp -j SNAT --to-source 86.107.168.15:62024-62535\n",
+		"-A REWEAVE-LB-POST -s 10.44.0.9/32 -o " + toW1 + " -m mark --mark 0x100000/0x100000 -p udp -j SNAT --to-source 86.107.168.15:62024-62535\n",
+		"-A REWEAVE-LB-POST -s 10.44.0.9/32 -o " + toW1 + " -m mark --mark 0x100000/0x100000 -p sctp -j SNAT --to-source 86.107.168.15:62024-62535\n",
+		// ping within the block too (the echo identifier is NAT's port for ICMP)
+		"-A REWEAVE-LB-POST -s 10.44.0.9/32 -o " + toW1 + " -m mark --mark 0x100000/0x100000 -p icmp -j SNAT --to-source 86.107.168.15:62024-62535\n",
+		// everything else (GRE, ...) from the address alone
+		"-A REWEAVE-LB-POST -s 10.44.0.9/32 -o " + toW1 + " -m mark --mark 0x100000/0x100000 -j SNAT --to-source 86.107.168.15\n",
+	} {
+		if !strings.Contains(r.Nat, want) {
+			t.Fatalf("member nat lacks %q:\n%s", want, r.Nat)
+		}
+	}
+	// the catch-all SNAT comes after the port-ranged ones (or ports would not stay in the block)
+	if strings.Index(r.Nat, "-j SNAT --to-source 86.107.168.15\n") < strings.Index(r.Nat, "-p icmp -j SNAT") {
+		t.Fatalf("SNAT order:\n%s", r.Nat)
+	}
+	for _, want := range []string{
+		"-A REWEAVE-LB-FWD -s 10.44.0.9/32 -o " + toW1 + " -m mark --mark 0x100000/0x100000 -j RETURN\n",
+		"-A REWEAVE-LB-FWD -s 10.44.0.9/32 -m mark --mark 0x100000/0x100000 -m comment --comment \"gitea/mail-0\" -j DROP\n",
+		"-A REWEAVE-LB-FWD -i rweg+ -d 86.107.168.15/32 -j DROP\n",
+		"-A REWEAVE-LB-FWD -i rweg+ -m conntrack --ctstate ESTABLISHED,RELATED -j RETURN\n",
+		"-A REWEAVE-LB-FWD -i rweg+ -j DROP\n",
+		"-A REWEAVE-LB-IN -p 4 -s 10.40.0.0/32 -m mac --mac-source aa:aa:aa:aa:aa:02 -j RETURN\n",
+		"-A REWEAVE-LB-IN -p 4 -j DROP\n",
+	} {
+		if !strings.Contains(r.Filter, want) {
+			t.Fatalf("member filter lacks %q:\n%s", want, r.Filter)
+		}
+	}
+	for _, want := range []string{
+		"-A REWEAVE-LB-MARK -m conntrack --ctdir ORIGINAL -m connmark --mark 0x100000/0x100000 -j MARK --set-xmark 0x100000/0x100000\n",
+		"-A REWEAVE-LB-MSS -o rweg+ -p tcp -m tcp --tcp-flags SYN,RST SYN -j TCPMSS --clamp-mss-to-pmtu\n",
+	} {
+		if !strings.Contains(r.Mangle, want) {
+			t.Fatalf("member mangle lacks %q:\n%s", want, r.Mangle)
+		}
+	}
+	if strings.Contains(r.Mangle, "--set-xmark 0x") && strings.Contains(r.Mangle, "-i "+toW1) {
+		t.Fatalf("a member's node does not tag what its holder sends back:\n%s", r.Mangle)
+	}
+	// the guard's DROP comes after the member's own RETURN: nothing else of its leaves
+	if strings.Index(r.Filter, toW1+" -m mark --mark 0x100000/0x100000 -j RETURN") > strings.Index(r.Filter, "\"gitea/mail-0\" -j DROP") {
+		t.Fatalf("guard order:\n%s", r.Filter)
+	}
+
+	// the holder: tags each tunnel's connections with its node, accepts each node's block, nothing foreign
+	h := ComputeEgress("w1", nodes, svcs, []*EgressService{e}, []*coreapi.Pod{pod("mail-1", "w1", "10.40.0.4", mailLabels)}, nil)
+	r = Render(Desired{}, h, "w1", podNet, nil)
+	idx := peerIndex(h, "w1")
+	tagW2 := fmt.Sprintf("0x%x", peerTag(idx["w2"]))
+	for _, want := range []string{
+		"-A REWEAVE-LB-MARK -m conntrack --ctdir REPLY -m connmark --mark 0x200000/0x200000 -j CONNMARK --restore-mark --nfmask 0xff200000 --ctmask 0xff200000\n",
+		"-A REWEAVE-LB-MARK -i " + toW2 + " -m conntrack --ctstate NEW -m connmark --mark 0x0/0xff200000 -m comment --comment \"from w2\" -j CONNMARK --set-xmark " + tagW2 + "/0xff200000\n",
+	} {
+		if !strings.Contains(r.Mangle, want) {
+			t.Fatalf("holder mangle lacks %q:\n%s", want, r.Mangle)
+		}
+	}
+	for _, want := range []string{
+		"-A REWEAVE-LB-FWD -i " + toW2 + " -m conntrack --ctdir ORIGINAL -m connmark ! --mark " + tagW2 + "/0xff200000 -j DROP\n",
+		"-A REWEAVE-LB-FWD -i " + toW2 + " -m conntrack --ctdir ORIGINAL -s 86.107.168.15/32 -p tcp -m tcp --sport 62024:62535 -m comment --comment \"block of w2\" -j RETURN\n",
+		"-A REWEAVE-LB-FWD -i " + toCP + " -m conntrack --ctdir ORIGINAL -s 86.107.168.15/32 -p sctp -m sctp --sport 61000:61511 -m comment --comment \"block of cp\" -j RETURN\n",
+		"-A REWEAVE-LB-FWD -i " + toW2 + " -m conntrack --ctdir ORIGINAL -s 86.107.168.15/32 -p udp -j DROP\n",
+		"-A REWEAVE-LB-FWD -i " + toW2 + " -m conntrack --ctdir ORIGINAL -s 86.107.168.15/32 -j RETURN\n",
+		"-A REWEAVE-LB-FWD -s 10.40.0.4/32 -o weave -m mark --mark 0x100000/0x100000 -j DROP\n",
+	} {
+		if !strings.Contains(r.Filter, want) {
+			t.Fatalf("holder filter lacks %q:\n%s", want, r.Filter)
+		}
+	}
+	// foreign tags are refused before anything of that tunnel is accepted; a port outside the block before the
+	// catch-all accept
+	if strings.Index(r.Filter, "! --mark "+tagW2) > strings.Index(r.Filter, "\"block of w2\"") ||
+		strings.Index(r.Filter, "-i "+toW2+" -m conntrack --ctdir ORIGINAL -s 86.107.168.15/32 -p udp -j DROP") >
+			strings.Index(r.Filter, "-i "+toW2+" -m conntrack --ctdir ORIGINAL -s 86.107.168.15/32 -j RETURN") {
+		t.Fatalf("holder filter order:\n%s", r.Filter)
+	}
+	if !strings.Contains(r.Nat, "-A REWEAVE-LB-POST -s 10.40.0.4/32 ! -o weave -m mark --mark 0x100000/0x100000 -p tcp -j SNAT --to-source 86.107.168.15:61512-62023\n") {
+		t.Fatalf("holder's own member:\n%s", r.Nat)
+	}
+	for _, line := range strings.Split(r.Nat+r.Filter, "\n") {
+		if strings.Contains(line, "10.44.0.10") {
+			t.Fatalf("a rule names a non-member: %s", line)
+		}
+	}
+}
+
+func TestTunnelNames(t *testing.T) {
+	a, b := TunnelName("reweave-worker-fe19b040"), TunnelName("reweave-cp-94a8db43")
+	if a == b || len(a) > 15 || !strings.HasPrefix(a, TunnelPrefix) || a != TunnelName("reweave-worker-fe19b040") {
+		t.Fatalf("tunnel names must be distinct, stable, short: %q %q", a, b)
+	}
+}
+
+func TestPolicyRules(t *testing.T) {
+	e := egressFor("mail", written())
+	svcs := []*coreapi.Service{mailSvc()}
+	h := ComputeEgress("w1", nodes, svcs, []*EgressService{e}, nil, nil)
+	rules := wantRules(h, "w1")
+	tables := peerTables(h, "w1")
+	idx := peerIndex(h, "w1")
+	ports, tags := 0, 0
+	for _, r := range rules {
+		if r.priority != HolderRulePriority || r.src != "" {
+			t.Fatalf("holder rule: %+v", r)
+		}
+		switch {
+		case r.ipproto != 0:
+			ports++
+			if r.dst != "86.107.168.15/32" || r.mark != 0 {
+				t.Fatalf("port rule: %+v", r)
+			}
+			if (r.from == 62024) != (r.table == tables["w2"]) || (r.from == 61000) != (r.table == tables["cp"]) {
+				t.Fatalf("block routed to the wrong node: %+v %v", r, tables)
+			}
+		default:
+			tags++
+			if r.mask != peerMask || r.dst != "" {
+				t.Fatalf("tag rule: %+v", r)
+			}
+			for name, i := range idx {
+				if r.mark == peerTag(i) && r.table != tables[name] {
+					t.Fatalf("tag of %s routed to table %d, want %d", name, r.table, tables[name])
+				}
+			}
+		}
+	}
+	if ports != 2*len(portProtocols) || tags != 2 {
+		t.Fatalf("2 peers: %d port rules (tcp, udp, sctp each), %d tag rules: %+v", ports, tags, rules)
+	}
+	if HolderRulePriority >= LocalRulePriority {
+		t.Fatalf("holder rules must precede the local table")
+	}
+	m := ComputeEgress("w2", nodes, svcs, []*EgressService{e}, []*coreapi.Pod{pod("mail-0", "w2", "10.44.0.9", mailLabels)}, nil)
+	rules = wantRules(m, "w2")
+	if len(rules) != 1 || rules[0].src != "10.44.0.9/32" || rules[0].mark != EgressMark || rules[0].mask != EgressMark ||
+		rules[0].table != peerTables(m, "w2")["w1"] {
+		t.Fatalf("member rule: %+v", rules)
+	}
+	u := ComputeEgress("w2", nodes, svcs, []*EgressService{egressFor("mail", EgressServiceStatus{})}, []*coreapi.Pod{pod("mail-0", "w2", "10.44.0.9", mailLabels)}, nil)
+	rules = wantRules(u, "w2")
+	if len(rules) != 1 || rules[0].table != EgressTableBase {
+		t.Fatalf("unreachable member must hit the unreachable table: %+v", rules)
+	}
+}

lb/model_test.go

@@ -204,9 +204,9 @@ func TestRenderedRules(t *testing.T) {
 	s := svc("gitea", "86.107.168.15")
 	sl := slice("gitea", map[string]bool{"10.40.0.7": true, "10.44.0.3": true, "10.44.0.4": true}, map[string]int32{"ssh": 2222, "https": 3000})
 	d := Compute("w1", nodes, []*coreapi.Service{s}, []*discovery.EndpointSlice{sl})
-	r := Render(d, podNet, []string{"REWEAVE-LBS-DEADBEEF0000"})
+	r := Render(d, EgressDesired{}, "w1", podNet, []string{"REWEAVE-LBS-DEADBEEF0000"})
 	for _, want := range []string{
-		"*nat\n:REWEAVE-LB - [0:0]\n:REWEAVE-LB-POST - [0:0]\n",
+		"*nat\n:REWEAVE-LB - [0:0]\n:REWEAVE-LB-POST - [0:0]\n:REWEAVE-LB-CLASS - [0:0]\n",
 		// the client's address survives weave's masquerade; a pod's own connection (hairpin) is masqueraded here
 		"-A REWEAVE-LB-POST -m conntrack --ctstate DNAT --ctorigdst 86.107.168.15/32 -s 10.32.0.0/12 -j MASQUERADE\n",
 		"-A REWEAVE-LB-POST -m conntrack --ctstate DNAT --ctorigdst 86.107.168.15/32 -j ACCEPT\n",
@@ -230,11 +230,11 @@ func TestRenderedRules(t *testing.T) {
 	if strings.Count(r.Nat, "MASQUERADE") != 1 || strings.Contains(r.Nat, "SNAT") {
 		t.Fatalf("only a pod's hairpin connection may be masqueraded:\n%s", r.Nat)
 	}
-	if r.Mangle != "*mangle\n:REWEAVE-LB-MARK - [0:0]\nCOMMIT\n" {
+	if r.Mangle != "*mangle\n:REWEAVE-LB-MARK - [0:0]\n:REWEAVE-LB-MSS - [0:0]\nCOMMIT\n" {
 		t.Fatalf("a holder marks nothing:\n%s", r.Mangle)
 	}
 	d = Compute("w2", nodes, []*coreapi.Service{s}, []*discovery.EndpointSlice{sl})
-	r = Render(d, podNet, nil)
+	r = Render(d, EgressDesired{}, "w2", podNet, nil)
 	m := d.Holders[0].Mark << 24
 	for _, want := range []string{
 		"-A REWEAVE-LB-MARK -m conntrack --ctdir REPLY -m connmark ! --mark 0x0/0xff000000 -j CONNMARK --restore-mark --nfmask 0xff000000 --ctmask 0xff000000\n",
@@ -244,7 +244,10 @@ func TestRenderedRules(t *testing.T) {
 			t.Fatalf("mangle lacks %q:\n%s", want, r.Mangle)
 		}
 	}
-	if r.Nat != "*nat\n:REWEAVE-LB - [0:0]\n:REWEAVE-LB-POST - [0:0]\nCOMMIT\n" {
+	if r.Nat != "*nat\n:REWEAVE-LB - [0:0]\n:REWEAVE-LB-POST - [0:0]\n:REWEAVE-LB-CLASS - [0:0]\nCOMMIT\n" {
 		t.Fatalf("a non-holder forwards nothing:\n%s", r.Nat)
 	}
+	if r.Filter != "*filter\n:REWEAVE-LB-FWD - [0:0]\n:REWEAVE-LB-IN - [0:0]\nCOMMIT\n" {
+		t.Fatalf("no EgressService, no filter rules:\n%s", r.Filter)
+	}
 }

net/portlock.go

@@ -0,0 +1,220 @@
+package net
+
+import (
+	"encoding/binary"
+	"errors"
+	"fmt"
+	"net"
+	"strings"
+
+	"github.com/vishvananda/netlink"
+	"github.com/vishvananda/netlink/nl"
+	"github.com/vishvananda/netns"
+	"golang.org/x/sys/unix"
+)
+
+// A pod's port is locked to the pod's own addresses: what the pod sends in through the host side of its veth has to
+// carry the pod's MAC as its Ethernet source and be either IPv4 from the pod's address or ARP speaking for exactly
+// that MAC and address (or probing, from 0.0.0.0). Everything else is dropped before the bridge sees it, so a pod can
+// claim neither another pod's identity nor a node's, on its own node or, through weave, on any other.
+//
+// The lock is a set of tc filters on the veth's ingress qdisc. They match the way iproute2 lays out "ether src"
+// (the Ethernet source at -8 and -4 from the network header) and "ip src" (at 12), and they go away with the veth.
+
+const (
+	portLockPrioIPv4  = 1
+	portLockPrioARP   = 2
+	portLockPrioProbe = 3
+	portLockPrioDrop  = 9
+)
+
+var ingressParent = netlink.MakeHandle(0xffff, 0)
+
+// PortLockFilters are the filters that lock the port with the given index to a pod's MAC and IPv4 address.
+func PortLockFilters(index int, mac net.HardwareAddr, ip net.IP) ([]netlink.Filter, error) {
+	if len(mac) != 6 {
+		return nil, fmt.Errorf("port lock needs an Ethernet MAC, got %q", mac)
+	}
+	ip4 := ip.To4()
+	if ip4 == nil {
+		return nil, fmt.Errorf("port lock needs an IPv4 address, got %q", ip)
+	}
+	src := etherSource(mac)
+	arp := func(spa net.IP) []netlink.TcU32Key {
+		keys := append([]netlink.TcU32Key{}, src...)
+		return append(keys,
+			// Ethernet/IPv4 ARP: hardware type 1, protocol 0x0800, address lengths 6 and 4 (any operation)
+			netlink.TcU32Key{Mask: 0xffffffff, Val: 0x00010800, Off: 0},
+			netlink.TcU32Key{Mask: 0xffff0000, Val: 0x06040000, Off: 4},
+			// the sender hardware address (8..13) and the sender protocol address (14..17)
+			netlink.TcU32Key{Mask: 0xffffffff, Val: binary.BigEndian.Uint32(mac[0:4]), Off: 8},
+			netlink.TcU32Key{Mask: 0xffffffff, Val: uint32(binary.BigEndian.Uint16(mac[4:6]))<<16 | uint32(binary.BigEndian.Uint16(spa[0:2])), Off: 12},
+			netlink.TcU32Key{Mask: 0xffff0000, Val: uint32(binary.BigEndian.Uint16(spa[2:4])) << 16, Off: 16},
+		)
+	}
+	ipv4 := append(append([]netlink.TcU32Key{}, src...), netlink.TcU32Key{Mask: 0xffffffff, Val: binary.BigEndian.Uint32(ip4), Off: 12})
+	return []netlink.Filter{
+		portLockU32(index, portLockPrioIPv4, unix.ETH_P_IP, ipv4),
+		portLockU32(index, portLockPrioARP, unix.ETH_P_ARP, arp(ip4)),
+		portLockU32(index, portLockPrioProbe, unix.ETH_P_ARP, arp(net.IPv4zero.To4())),
+		&netlink.MatchAll{
+			FilterAttrs: netlink.FilterAttrs{LinkIndex: index, Parent: ingressParent, Priority: portLockPrioDrop, Protocol: unix.ETH_P_ALL},
+			Actions:     []netlink.Action{gact(netlink.TC_ACT_SHOT)},
+		},
+	}, nil
+}
+
+// etherSource matches the Ethernet source address, which sits 8 bytes before the network header.
+func etherSource(mac net.HardwareAddr) []netlink.TcU32Key {
+	return []netlink.TcU32Key{
+		{Mask: 0xffffffff, Val: binary.BigEndian.Uint32(mac[0:4]), Off: -8},
+		{Mask: 0xffff0000, Val: uint32(binary.BigEndian.Uint16(mac[4:6])) << 16, Off: -4},
+	}
+}
+
+func portLockU32(index int, prio uint16, proto uint16, keys []netlink.TcU32Key) *netlink.U32 {
+	// clipped: the vendored netlink serialises cap(Keys) keys, and spare capacity would go out as extra all-zero keys
+	keys = keys[:len(keys):len(keys)]
+	return &netlink.U32{
+		FilterAttrs: netlink.FilterAttrs{LinkIndex: index, Parent: ingressParent, Priority: prio, Protocol: proto},
+		Sel:         &netlink.TcU32Sel{Flags: nl.TC_U32_TERMINAL, Keys: keys},
+		Actions:     []netlink.Action{gact(netlink.TC_ACT_OK)},
+	}
+}
+
+func gact(action netlink.TcAct) netlink.Action {
+	return &netlink.GenericAction{ActionAttrs: netlink.ActionAttrs{Action: action}}
+}
+
+func ingressQdisc(index int) *netlink.Ingress {
+	return &netlink.Ingress{QdiscAttrs: netlink.QdiscAttrs{LinkIndex: index, Handle: ingressParent, Parent: netlink.HANDLE_INGRESS}}
+}
+
+// LockPort locks a pod's port (the host side of its veth) to the pod's MAC and IPv4 address. A port already locked
+// to them is left alone; anything else on its ingress is replaced, and the pod's own traffic is let through before
+// the rest is dropped, so a running pod never loses its own frames.
+func LockPort(link netlink.Link, mac net.HardwareAddr, ip net.IP) error {
+	index := link.Attrs().Index
+	want, err := PortLockFilters(index, mac, ip)
+	if err != nil {
+		return err
+	}
+	if locked, _ := portHasFilters(link, want); locked {
+		return nil
+	}
+	// removing the ingress qdisc takes its filters with it
+	if err := netlink.QdiscDel(ingressQdisc(index)); err != nil && !errors.Is(err, unix.ENOENT) && !errors.Is(err, unix.EINVAL) {
+		return fmt.Errorf("clearing ingress of %s: %w", link.Attrs().Name, err)
+	}
+	if err := netlink.QdiscAdd(ingressQdisc(index)); err != nil {
+		return fmt.Errorf("adding ingress qdisc to %s: %w", link.Attrs().Name, err)
+	}
+	for _, f := range want {
+		if err := netlink.FilterAdd(f); err != nil {
+			return fmt.Errorf("adding port lock filter (prio %d) to %s: %w", f.Attrs().Priority, link.Attrs().Name, err)
+		}
+	}
+	return nil
+}
+
+// LockContainerPort locks the port of a pod attached by the CNI plugin: the host side of the veth whose container
+// end is ifName in ns, to that end's MAC and the pod's address.
+func LockContainerPort(ns netns.NsHandle, ifName string, ip net.IP) error {
+	var mac net.HardwareAddr
+	peer := 0
+	if err := WithNetNSLink(ns, ifName, func(link netlink.Link) error {
+		// a veth's link attribute is its peer's index, in the peer's namespace: the host's
+		mac, peer = link.Attrs().HardwareAddr, link.Attrs().ParentIndex
+		return nil
+	}); err != nil {
+		return err
+	}
+	if peer == 0 {
+		return fmt.Errorf("%s has no peer: not a veth", ifName)
+	}
+	host, err := netlink.LinkByIndex(peer)
+	if err != nil {
+		return fmt.Errorf("host side of %s: %w", ifName, err)
+	}
+	if host.Type() != "veth" || !strings.HasPrefix(host.Attrs().Name, vethPrefix+"pl") {
+		return fmt.Errorf("host side of %s is %s (%s), not a weave pod veth", ifName, host.Attrs().Name, host.Type())
+	}
+	return LockPort(host, mac, ip)
+}
+
+// PortLocked reports whether a port is locked to exactly this MAC and address.
+func PortLocked(link netlink.Link, mac net.HardwareAddr, ip net.IP) (bool, error) {
+	want, err := PortLockFilters(link.Attrs().Index, mac, ip)
+	if err != nil {
+		return false, err
+	}
+	return portHasFilters(link, want)
+}
+
+func portHasFilters(link netlink.Link, want []netlink.Filter) (bool, error) {
+	have, err := netlink.FilterList(link, ingressParent)
+	if err != nil {
+		return false, err
+	}
+	var got []netlink.Filter
+	for _, f := range have {
+		// u32 lists the hash table that holds a filter as a filter of its own, without a selector
+		if u, ok := f.(*netlink.U32); ok && u.Sel == nil {
+			continue
+		}
+		got = append(got, f)
+	}
+	if len(got) != len(want) {
+		return false, nil
+	}
+	for _, w := range want {
+		found := false
+		for _, g := range got {
+			if sameFilter(g, w) {
+				found = true
+				break
+			}
+		}
+		if !found {
+			return false, nil
+		}
+	}
+	return true, nil
+}
+
+func sameFilter(a, b netlink.Filter) bool {
+	if a.Attrs().Priority != b.Attrs().Priority || a.Attrs().Protocol != b.Attrs().Protocol || a.Type() != b.Type() {
+		return false
+	}
+	switch fa := a.(type) {
+	case *netlink.U32:
+		fb := b.(*netlink.U32)
+		if fa.Sel == nil || fb.Sel == nil || len(fa.Sel.Keys) != len(fb.Sel.Keys) || fa.Sel.Flags&nl.TC_U32_TERMINAL == 0 {
+			return false
+		}
+		for i := range fa.Sel.Keys {
+			ka, kb := fa.Sel.Keys[i], fb.Sel.Keys[i]
+			if ka.Mask != kb.Mask || ka.Val != kb.Val || ka.Off != kb.Off {
+				return false
+			}
+		}
+		return sameActions(fa.Actions, fb.Actions)
+	case *netlink.MatchAll:
+		return sameActions(fa.Actions, b.(*netlink.MatchAll).Actions)
+	}
+	return false
+}
+
+func sameActions(a, b []netlink.Action) bool {
+	if len(a) != len(b) {
+		return false
+	}
+	for i := range a {
+		ga, ok1 := a[i].(*netlink.GenericAction)
+		gb, ok2 := b[i].(*netlink.GenericAction)
+		if !ok1 || !ok2 || ga.Attrs().Action != gb.Attrs().Action {
+			return false
+		}
+	}
+	return true
+}

net/portlock_kernel_test.go

@@ -0,0 +1,228 @@
+package net
+
+import (
+	"bytes"
+	"encoding/binary"
+	"net"
+	"os"
+	"runtime"
+	"testing"
+	"time"
+
+	"github.com/vishvananda/netlink"
+	"github.com/vishvananda/netns"
+	"golang.org/x/sys/unix"
+)
+
+// TestPortLockInKernel builds a bridge with two pods in throwaway network namespaces, locks pod A's port and sends
+// frames from A: what A may send reaches B, what claims anything but A's own addresses does not. It needs root.
+func TestPortLockInKernel(t *testing.T) {
+	if os.Geteuid() != 0 {
+		t.Skip("needs root (creates network namespaces)")
+	}
+	runtime.LockOSThread()
+	defer runtime.UnlockOSThread()
+	orig, err := netns.Get()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer orig.Close()
+	defer netns.Set(orig)
+
+	hub, err := netns.New()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer hub.Close()
+	podA, _ := netns.New()
+	defer podA.Close()
+	podB, _ := netns.New()
+	defer podB.Close()
+	must(t, netns.Set(hub))
+
+	br := &netlink.Bridge{LinkAttrs: netlink.LinkAttrs{Name: "br0"}}
+	must(t, netlink.LinkAdd(br))
+	must(t, netlink.LinkSetUp(br))
+	ipA, ipB := net.IPv4(10, 99, 0, 1).To4(), net.IPv4(10, 99, 0, 2).To4()
+	hostA, macA := attach(t, hub, podA, br, "vethwepltsta", ipA)
+	_, macB := attach(t, hub, podB, br, "vethwepltstb", ipB)
+
+	// as the CNI plugin does it: from the pod's end of the veth, in the pod's namespace
+	must(t, LockContainerPort(podA, "eth0", ipA))
+	if ok, err := PortLocked(hostA, macA, ipA); err != nil || !ok {
+		dumpIngress(t, hostA)
+		t.Fatalf("port not locked after LockContainerPort: %v %v", ok, err)
+	}
+	// locking again changes nothing (the filters are compared, not rebuilt)
+	must(t, LockPort(hostA, macA, ipA))
+	if ok, _ := PortLocked(hostA, macA, ipA); !ok {
+		t.Fatal("port lost its lock on a repeated LockPort")
+	}
+
+	send := socketIn(t, podA, "eth0")
+	recv := socketIn(t, podB, "eth0")
+	defer unix.Close(send)
+	defer unix.Close(recv)
+	other, _ := net.ParseMAC("02:00:00:00:00:99")
+	spoofIP := net.IPv4(10, 99, 0, 200).To4()
+
+	cases := []struct {
+		name   string
+		frame  []byte
+		passes bool
+	}{
+		{"ipv4 from its own addresses", ipv4Frame(macB, macA, ipA, ipB, "own-ip"), true},
+		{"ipv4 from another MAC", ipv4Frame(macB, other, ipA, ipB, "mac-spoof"), false},
+		{"ipv4 from another address", ipv4Frame(macB, macA, spoofIP, ipB, "ip-spoof"), false},
+		{"arp for itself", arpFrame(macA, macA, ipA, ipB, "own-arp"), true},
+		{"arp probe", arpFrame(macA, macA, net.IPv4zero.To4(), ipB, "probe"), true},
+		{"arp claiming another address", arpFrame(macA, macA, spoofIP, ipB, "arp-ip-spoof"), false},
+		{"arp claiming another MAC", arpFrame(macA, other, ipA, ipB, "arp-sha-spoof"), false},
+		{"arp from another MAC", arpFrame(other, other, ipA, ipB, "arp-eth-spoof"), false},
+		{"ipv6", etherFrame(macB, macA, 0x86dd, []byte("ipv6-frame")), false},
+		{"other ethertype", etherFrame(macB, macA, 0x88b5, []byte("experimental")), false},
+	}
+	for _, c := range cases {
+		drain(recv)
+		if _, err := unix.Write(send, c.frame); err != nil {
+			t.Fatalf("%s: send: %v", c.name, err)
+		}
+		got := sawPayload(recv, c.frame[len(c.frame)-12:], 300*time.Millisecond)
+		if got != c.passes {
+			t.Errorf("%s: reached B = %v, want %v", c.name, got, c.passes)
+		}
+	}
+}
+
+func dumpIngress(t *testing.T, link netlink.Link) {
+	have, _ := netlink.FilterList(link, netlink.MakeHandle(0xffff, 0))
+	for _, f := range have {
+		t.Logf("have %s %+v", f.Type(), f.Attrs())
+		if u, ok := f.(*netlink.U32); ok && u.Sel != nil {
+			t.Logf("  sel flags %x keys %+v actions %+v", u.Sel.Flags, u.Sel.Keys, u.Actions)
+		}
+		if m, ok := f.(*netlink.MatchAll); ok {
+			t.Logf("  actions %+v", m.Actions)
+		}
+	}
+}
+
+func must(t *testing.T, err error) {
+	t.Helper()
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
+// attach plugs a pod namespace into the bridge through a veth named host on the hub side, eth0 on the pod side.
+func attach(t *testing.T, hub, pod netns.NsHandle, br netlink.Link, host string, ip net.IP) (netlink.Link, net.HardwareAddr) {
+	t.Helper()
+	must(t, netlink.LinkAdd(&netlink.Veth{LinkAttrs: netlink.LinkAttrs{Name: host}, PeerName: "tmp" + host[len(host)-2:]}))
+	h, err := netlink.LinkByName(host)
+	must(t, err)
+	must(t, netlink.LinkSetMaster(h, br))
+	must(t, netlink.LinkSetUp(h))
+	p, err := netlink.LinkByName("tmp" + host[len(host)-2:])
+	must(t, err)
+	must(t, netlink.LinkSetNsFd(p, int(pod)))
+	must(t, netns.Set(pod))
+	p, err = netlink.LinkByName("tmp" + host[len(host)-2:])
+	must(t, err)
+	must(t, netlink.LinkSetName(p, "eth0"))
+	must(t, netlink.AddrAdd(p, &netlink.Addr{IPNet: &net.IPNet{IP: ip, Mask: net.CIDRMask(24, 32)}}))
+	must(t, netlink.LinkSetUp(p))
+	mac := p.Attrs().HardwareAddr
+	must(t, netns.Set(hub))
+	return h, mac
+}
+
+// socketIn opens a raw packet socket on an interface of a namespace (the socket stays in that namespace).
+func socketIn(t *testing.T, ns netns.NsHandle, ifName string) int {
+	t.Helper()
+	cur, _ := netns.Get()
+	defer cur.Close()
+	must(t, netns.Set(ns))
+	defer netns.Set(cur)
+	link, err := netlink.LinkByName(ifName)
+	must(t, err)
+	fd, err := unix.Socket(unix.AF_PACKET, unix.SOCK_RAW, int(htons(unix.ETH_P_ALL)))
+	must(t, err)
+	must(t, unix.Bind(fd, &unix.SockaddrLinklayer{Protocol: htons(unix.ETH_P_ALL), Ifindex: link.Attrs().Index}))
+	return fd
+}
+
+func htons(v uint16) uint16 { return v<<8 | v>>8 }
+
+func drain(fd int) {
+	buf := make([]byte, 2048)
+	for {
+		if _, _, err := unix.Recvfrom(fd, buf, unix.MSG_DONTWAIT); err != nil {
+			return
+		}
+	}
+}
+
+func sawPayload(fd int, marker []byte, within time.Duration) bool {
+	buf := make([]byte, 2048)
+	deadline := time.Now().Add(within)
+	for time.Now().Before(deadline) {
+		n, _, err := unix.Recvfrom(fd, buf, unix.MSG_DONTWAIT)
+		if err != nil {
+			time.Sleep(5 * time.Millisecond)
+			continue
+		}
+		if bytes.Contains(buf[:n], marker) {
+			return true
+		}
+	}
+	return false
+}
+
+func etherFrame(dst, src net.HardwareAddr, ethertype uint16, payload []byte) []byte {
+	f := make([]byte, 14, 14+len(payload)+12)
+	copy(f[0:6], dst)
+	copy(f[6:12], src)
+	binary.BigEndian.PutUint16(f[12:14], ethertype)
+	f = append(f, payload...)
+	// a fixed-size tail the receiver looks for, so each case is told apart
+	tail := make([]byte, 12)
+	copy(tail, payload)
+	return append(f, tail...)
+}
+
+func ipv4Frame(dst, src net.HardwareAddr, from, to net.IP, tag string) []byte {
+	body := []byte(tag)
+	h := make([]byte, 20)
+	h[0] = 0x45
+	binary.BigEndian.PutUint16(h[2:4], uint16(20+len(body)+12))
+	h[8] = 64
+	h[9] = 253 // experimental protocol: nothing answers it
+	copy(h[12:16], from)
+	copy(h[16:20], to)
+	binary.BigEndian.PutUint16(h[10:12], checksum(h))
+	return etherFrame(dst, src, unix.ETH_P_IP, append(h, body...))
+}
+
+func arpFrame(ethSrc, sha net.HardwareAddr, spa, tpa net.IP, tag string) []byte {
+	a := make([]byte, 28)
+	binary.BigEndian.PutUint16(a[0:2], 1)
+	binary.BigEndian.PutUint16(a[2:4], unix.ETH_P_IP)
+	a[4], a[5] = 6, 4
+	binary.BigEndian.PutUint16(a[6:8], 1) // request
+	copy(a[8:14], sha)
+	copy(a[14:18], spa)
+	copy(a[24:28], tpa)
+	bcast, _ := net.ParseMAC("ff:ff:ff:ff:ff:ff")
+	return etherFrame(bcast, ethSrc, unix.ETH_P_ARP, append(a, []byte(tag)...))
+}
+
+func checksum(b []byte) uint16 {
+	var s uint32
+	for i := 0; i+1 < len(b); i += 2 {
+		s += uint32(binary.BigEndian.Uint16(b[i:]))
+	}
+	for s>>16 != 0 {
+		s = s&0xffff + s>>16
+	}
+	return ^uint16(s)
+}

net/portlock_test.go

@@ -0,0 +1,115 @@
+package net
+
+import (
+	"net"
+	"testing"
+
+	"github.com/vishvananda/netlink"
+	"github.com/vishvananda/netlink/nl"
+	"golang.org/x/sys/unix"
+)
+
+// The keys iproute2 writes for "u32 match ether src 02:11:22:33:44:55 match ip src 10.32.0.7/32" (tc filter show,
+// iproute2 6.x): the port lock has to match the same bytes.
+func TestPortLockMatchesIproute2Layout(t *testing.T) {
+	mac, _ := net.ParseMAC("02:11:22:33:44:55")
+	fs, err := PortLockFilters(7, mac, net.ParseIP("10.32.0.7"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(fs) != 4 {
+		t.Fatalf("want 4 filters, got %d", len(fs))
+	}
+	ipv4 := fs[0].(*netlink.U32)
+	want := []netlink.TcU32Key{
+		{Val: 0x02112233, Mask: 0xffffffff, Off: -8},
+		{Val: 0x44550000, Mask: 0xffff0000, Off: -4},
+		{Val: 0x0a200007, Mask: 0xffffffff, Off: 12},
+	}
+	if ipv4.Protocol != unix.ETH_P_IP || ipv4.Priority != portLockPrioIPv4 || ipv4.Sel.Flags&nl.TC_U32_TERMINAL == 0 {
+		t.Fatalf("ipv4 filter attrs: %+v flags %x", ipv4.FilterAttrs, ipv4.Sel.Flags)
+	}
+	assertKeys(t, "ipv4", ipv4.Sel.Keys, want)
+	for _, f := range fs {
+		if f.Attrs().LinkIndex != 7 || f.Attrs().Parent != netlink.MakeHandle(0xffff, 0) {
+			t.Fatalf("filter not on ingress of link 7: %+v", f.Attrs())
+		}
+	}
+}
+
+func TestPortLockARP(t *testing.T) {
+	mac, _ := net.ParseMAC("02:11:22:33:44:55")
+	fs, _ := PortLockFilters(7, mac, net.ParseIP("10.32.0.7"))
+	common := []netlink.TcU32Key{
+		{Val: 0x02112233, Mask: 0xffffffff, Off: -8},
+		{Val: 0x44550000, Mask: 0xffff0000, Off: -4},
+		{Val: 0x00010800, Mask: 0xffffffff, Off: 0},
+		{Val: 0x06040000, Mask: 0xffff0000, Off: 4},
+		{Val: 0x02112233, Mask: 0xffffffff, Off: 8},
+	}
+	own := append(append([]netlink.TcU32Key{}, common...),
+		netlink.TcU32Key{Val: 0x44550a20, Mask: 0xffffffff, Off: 12}, // sha 44:55, spa 10.32
+		netlink.TcU32Key{Val: 0x00070000, Mask: 0xffff0000, Off: 16}, // spa .0.7
+	)
+	probe := append(append([]netlink.TcU32Key{}, common...),
+		netlink.TcU32Key{Val: 0x44550000, Mask: 0xffffffff, Off: 12},
+		netlink.TcU32Key{Val: 0x00000000, Mask: 0xffff0000, Off: 16},
+	)
+	arp, pr := fs[1].(*netlink.U32), fs[2].(*netlink.U32)
+	if arp.Protocol != unix.ETH_P_ARP || pr.Protocol != unix.ETH_P_ARP {
+		t.Fatal("ARP filters must be protocol arp")
+	}
+	assertKeys(t, "arp", arp.Sel.Keys, own)
+	assertKeys(t, "probe", pr.Sel.Keys, probe)
+	drop := fs[3].(*netlink.MatchAll)
+	if drop.Protocol != unix.ETH_P_ALL || drop.Priority != portLockPrioDrop {
+		t.Fatalf("drop filter attrs: %+v", drop.FilterAttrs)
+	}
+	if a := drop.Actions[0].(*netlink.GenericAction); a.Attrs().Action != netlink.TC_ACT_SHOT {
+		t.Fatalf("last filter must drop, has action %v", a.Attrs().Action)
+	}
+	for _, f := range fs[:3] {
+		if a := f.(*netlink.U32).Actions[0].(*netlink.GenericAction); a.Attrs().Action != netlink.TC_ACT_OK {
+			t.Fatalf("prio %d must pass, has action %v", f.Attrs().Priority, a.Attrs().Action)
+		}
+	}
+}
+
+func TestPortLockRefusesWhatItCannotPin(t *testing.T) {
+	mac, _ := net.ParseMAC("02:11:22:33:44:55")
+	if _, err := PortLockFilters(7, mac, net.ParseIP("fd00::1")); err == nil {
+		t.Fatal("an IPv6 address must be refused: the lock pins IPv4 and ARP")
+	}
+	long, _ := net.ParseMAC("02:11:22:33:44:55:66:77")
+	if _, err := PortLockFilters(7, long, net.ParseIP("10.32.0.7")); err == nil {
+		t.Fatal("a non-Ethernet hardware address must be refused")
+	}
+}
+
+func TestSameFilterComparesKeysAndActions(t *testing.T) {
+	mac, _ := net.ParseMAC("02:11:22:33:44:55")
+	a, _ := PortLockFilters(7, mac, net.ParseIP("10.32.0.7"))
+	b, _ := PortLockFilters(7, mac, net.ParseIP("10.32.0.8"))
+	if !sameFilter(a[0], a[0]) || sameFilter(a[0], b[0]) {
+		t.Fatal("ipv4 filters for different addresses must differ")
+	}
+	if !sameFilter(a[3], b[3]) {
+		t.Fatal("the drop filters are the same for every pod")
+	}
+	if sameFilter(a[1], a[2]) {
+		t.Fatal("the ARP and probe filters differ in priority and keys")
+	}
+}
+
+func assertKeys(t *testing.T, what string, got, want []netlink.TcU32Key) {
+	t.Helper()
+	if len(got) != len(want) {
+		t.Fatalf("%s: want %d keys, got %d: %+v", what, len(want), len(got), got)
+	}
+	for i := range want {
+		if got[i].Val != want[i].Val || got[i].Mask != want[i].Mask || got[i].Off != want[i].Off {
+			t.Fatalf("%s key %d: want %08x/%08x at %d, got %08x/%08x at %d", what, i,
+				want[i].Val, want[i].Mask, want[i].Off, got[i].Val, got[i].Mask, got[i].Off)
+		}
+	}
+}

plugin/net/cni.go

@@ -1,6 +1,7 @@
 package plugin
 
 import (
+	"context"
 	"crypto/rand"
 	"encoding/json"
 	"errors"
@@ -14,6 +15,7 @@ import (
 	current "github.com/containernetworking/cni/pkg/types/100"
 	"github.com/containernetworking/plugins/pkg/ipam"
 	weaveapi "github.com/rajch/weave/api"
+	"github.com/rajch/weave/lb/admit"
 	weavenet "github.com/rajch/weave/net"
 	ipamplugin "github.com/rajch/weave/plugin/ipam"
 	"github.com/vishvananda/netlink"
@@ -86,6 +88,18 @@ func (c *CNIPlugin) CmdAdd(args *skel.CmdArgs) error {
 	// Only expecting one address
 	ip := result.IPs[0]
 
+	// An EgressService member's first packet must already leave from its Service's address (or not at all):
+	// weave-npc installs the pod's egress rules before the pod has an interface. No answer, no pod (the
+	// runtime calls DEL, which releases the address, and kubelet retries).
+	if conf.EgressAdmission {
+		if ns, name := admit.PodFromArgs(args.Args); ns != "" && name != "" {
+			if _, err := admit.Admit(context.Background(), admit.SocketPath,
+				admit.Request{Namespace: ns, Name: name, IP: ip.Address.IP.String()}); err != nil {
+				return err
+			}
+		}
+	}
+
 	// If config says nothing about routes or gateway, default one will be via the bridge
 	if len(result.Routes) == 0 && ip.Gateway == nil {
 		bridgeIP, err := weavenet.FindBridgeIP(conf.BrName, &ip.Address)
@@ -147,6 +161,11 @@ func (c *CNIPlugin) CmdAdd(args *skel.CmdArgs) error {
 	}); err != nil {
 		return fmt.Errorf("error setting up routes: %s", err)
 	}
+	// the pod may only speak as itself: its port drops whatever claims another MAC or address. The pod has not
+	// started yet; if its port cannot be locked, it does not start.
+	if err := weavenet.LockContainerPort(ns, args.IfName, ip.Address.IP); err != nil {
+		return fmt.Errorf("error locking the pod's port: %s", err)
+	}
 
 	result.DNS = conf.DNS
 	return types.PrintResult(result, conf.CNIVersion)
@@ -241,4 +260,6 @@ type NetConf struct {
 	IPMasq      bool   `json:"ipMasq"`
 	MTU         int    `json:"mtu"`
 	HairpinMode bool   `json:"hairpinMode"`
+	// EgressAdmission: weave-npc must admit every pod before it gets its interface (EgressService members)
+	EgressAdmission bool `json:"egressAdmission"`
 }

prog/weave-kube/init.sh

@@ -63,4 +63,9 @@ if ! mkdir -p $HOST_ROOT/opt/cni/bin ; then
 fi
 mkdir -p $HOST_ROOT/etc/cni/net.d
 export HOST_ROOT
-/home/weave/weave --local setup-cni
+/home/weave/weave --local setup-cni
+
+# Lock the ports of pods attached before the CNI plugin locked ports itself (it locks every pod it attaches from
+# now on), from the runtime's record of each attachment. A port that cannot be locked stays as it was and is
+# reported; weave starts regardless, since holding back the whole node's network would not make that port safer.
+/usr/bin/weaveutil lock-ports "$HOST_ROOT/var/lib/cni/results" weave || echo "Some pods' ports could not be locked (see above)" >&2

prog/weave-kube/weave-daemonset-k8s-1.11.yaml

@@ -1,6 +1,68 @@
 apiVersion: v1
 kind: List
 items:
+  # EgressService (reweave.levk.dev/v1alpha1), served by weave-npc's lb agent: a LoadBalancer Service's pods
+  # leave the cluster from the Service's address (named after the Service, same namespace)
+  - apiVersion: apiextensions.k8s.io/v1
+    kind: CustomResourceDefinition
+    metadata:
+      name: egressservices.reweave.levk.dev
+      labels:
+        name: weave-net
+    spec:
+      group: reweave.levk.dev
+      scope: Namespaced
+      names:
+        plural: egressservices
+        singular: egressservice
+        kind: EgressService
+        listKind: EgressServiceList
+      versions:
+        - name: v1alpha1
+          served: true
+          storage: true
+          subresources:
+            status: {}
+          additionalPrinterColumns:
+            - {name: Emit, type: string, jsonPath: .spec.emit}
+            - {name: Host, type: string, jsonPath: .status.host}
+            - {name: Address, type: string, jsonPath: .status.address}
+            - {name: Age, type: date, jsonPath: .metadata.creationTimestamp}
+          schema:
+            openAPIV3Schema:
+              type: object
+              description: >-
+                Makes every pod the same-named LoadBalancer Service (class reweave.levk.dev/node-address)
+                selects leave the cluster from that Service's address, or not at all.
+              required: [spec]
+              properties:
+                spec:
+                  type: object
+                  required: [emit]
+                  properties:
+                    sourceIPBy:
+                      type: string
+                      description: LoadBalancerIP - members leave from the Service's load-balancer address.
+                      enum: [LoadBalancerIP]
+                      default: LoadBalancerIP
+                    emit:
+                      type: string
+                      description: Holder - members' packets are tunnelled to the node holding the address, which emits them.
+                      enum: [Holder]
+                status:
+                  type: object
+                  properties:
+                    host: {type: string}
+                    address: {type: string}
+                    blocks:
+                      type: array
+                      items:
+                        type: object
+                        required: [node, from, to]
+                        properties:
+                          node: {type: string}
+                          from: {type: integer, format: int32}
+                          to: {type: integer, format: int32}
   - apiVersion: v1
     kind: ServiceAccount
     metadata:
@@ -78,6 +140,20 @@ items:
         - nodes
         verbs:
         - patch
+      - apiGroups:
+        - reweave.levk.dev
+        resources:
+        - egressservices
+        verbs:
+        - get
+        - list
+        - watch
+      - apiGroups:
+        - reweave.levk.dev
+        resources:
+        - egressservices/status
+        verbs:
+        - patch
   - apiVersion: rbac.authorization.k8s.io/v1
     kind: ClusterRoleBinding
     metadata:
@@ -169,6 +245,15 @@ items:
                 - name: xtables-lock
                   mountPath: /run/xtables.lock
                   readOnly: false
+                # the runtime's record of each pod attachment, and the pods' network namespaces it names: init
+                # locks the ports of pods attached before the CNI plugin locked ports itself
+                - name: cni-cache
+                  mountPath: /host/var/lib/cni
+                  readOnly: true
+                - name: netns
+                  mountPath: /var/run/netns
+                  mountPropagation: HostToContainer
+                  readOnly: true
           containers:
             - name: weave
               command:
@@ -226,6 +311,8 @@ items:
                 - name: xtables-lock
                   mountPath: /run/xtables.lock
                   readOnly: false
+                - name: weave-npc-run
+                  mountPath: /run/weave-npc
           hostNetwork: true
           dnsPolicy: ClusterFirstWithHostNet
           hostPID: false
@@ -264,6 +351,18 @@ items:
               hostPath:
                 path: /run/xtables.lock
                 type: FileOrCreate
+            - name: weave-npc-run
+              hostPath:
+                path: /run/weave-npc
+                type: DirectoryOrCreate
+            - name: cni-cache
+              hostPath:
+                path: /var/lib/cni
+                type: DirectoryOrCreate
+            - name: netns
+              hostPath:
+                path: /var/run/netns
+                type: DirectoryOrCreate
           priorityClassName: system-node-critical
       updateStrategy:
         type: RollingUpdate

prog/weave-npc/main.go

@@ -340,8 +340,11 @@ func root(cmd *cobra.Command, args []string) {
 	go npController.Run(wait.NeverStop)
 
 	if serveLB {
-		// LoadBalancer Services of class lb.Class: addresses this node holds, client address kept
-		go lb.NewController(nodeName, client, lb.NewDatapath(ipt)).Run(context.Background())
+		// LoadBalancer Services of class lb.Class: addresses this node holds, client address kept; and
+		// EgressServices: those Services' pods leave the cluster from that address (or not at all)
+		egress, err := lb.NewEgressClient(config)
+		handleFatal(err)
+		go lb.NewController(nodeName, client, egress, lb.NewDatapath(ipt)).Run(context.Background())
 	}
 
 	signals := make(chan os.Signal, 1)

prog/weaveutil/main.go

@@ -42,6 +42,7 @@ func init() {
 		"rewrite-etc-hosts":        rewriteEtcHosts,
 		"get-db-flag":              getDBFlag,
 		"set-db-flag":              setDBFlag,
+		"lock-ports":               lockPorts,
 	}
 }
 

prog/weaveutil/portlock.go

@@ -0,0 +1,114 @@
+package main
+
+import (
+	"encoding/json"
+	"fmt"
+	"net"
+	"os"
+	"path/filepath"
+	"strings"
+
+	"github.com/vishvananda/netns"
+
+	weavenet "github.com/rajch/weave/net"
+)
+
+// lockPorts locks the ports of pods that were attached before the CNI plugin locked ports itself (it locks every
+// pod it attaches). It works from the runtime's own record of each attachment, libcni's result cache: the pod's
+// network namespace, the interface the plugin set up in it and the address it gave it. Ports already locked are
+// left alone; pods whose namespace is gone are skipped.
+func lockPorts(args []string) error {
+	if len(args) != 2 {
+		cmdUsage("lock-ports", "<cni-result-cache-dir> <network-name>")
+	}
+	dir, network := args[0], args[1]
+	entries, err := os.ReadDir(dir)
+	if os.IsNotExist(err) {
+		return nil // nothing attached yet
+	}
+	if err != nil {
+		return err
+	}
+	var locked, gone, failed int
+	for _, e := range entries {
+		if e.IsDir() || !strings.HasPrefix(e.Name(), network+"-") {
+			continue
+		}
+		data, err := os.ReadFile(filepath.Join(dir, e.Name()))
+		if err != nil {
+			fmt.Fprintf(os.Stderr, "lock-ports: %s: %v\n", e.Name(), err)
+			failed++
+			continue
+		}
+		a, err := parseAttachment(data)
+		if err != nil {
+			fmt.Fprintf(os.Stderr, "lock-ports: %s: %v\n", e.Name(), err)
+			failed++
+			continue
+		}
+		if a.network != network {
+			continue
+		}
+		ns, err := netns.GetFromPath(a.netns)
+		if os.IsNotExist(err) {
+			gone++ // the pod is gone; the runtime has yet to clear its record
+			continue
+		}
+		if err != nil {
+			fmt.Fprintf(os.Stderr, "lock-ports: %s: namespace %s: %v\n", a.containerID, a.netns, err)
+			failed++
+			continue
+		}
+		err = weavenet.LockContainerPort(ns, a.ifName, a.ip)
+		ns.Close()
+		if err != nil {
+			fmt.Fprintf(os.Stderr, "lock-ports: %s (%s): %v\n", a.containerID, a.ip, err)
+			failed++
+			continue
+		}
+		locked++
+	}
+	fmt.Printf("lock-ports: %d locked, %d gone, %d failed\n", locked, gone, failed)
+	if failed > 0 {
+		return fmt.Errorf("lock-ports: %d pods' ports not locked", failed)
+	}
+	return nil
+}
+
+type attachment struct {
+	network, containerID, netns, ifName string
+	ip                                  net.IP
+}
+
+// parseAttachment reads one libcni cache entry ("cniCacheV1"): which network attached which container, in which
+// namespace, through which interface, with which IPv4 address.
+func parseAttachment(data []byte) (attachment, error) {
+	var c struct {
+		Kind        string `json:"kind"`
+		ContainerID string `json:"containerId"`
+		IfName      string `json:"ifName"`
+		NetworkName string `json:"networkName"`
+		NetNS       string `json:"netns"`
+		Result      struct {
+			IPs []struct {
+				Address string `json:"address"`
+			} `json:"ips"`
+		} `json:"result"`
+	}
+	if err := json.Unmarshal(data, &c); err != nil {
+		return attachment{}, err
+	}
+	if c.Kind != "cniCacheV1" {
+		return attachment{}, fmt.Errorf("unknown cache entry kind %q", c.Kind)
+	}
+	if c.NetNS == "" || c.IfName == "" {
+		return attachment{}, fmt.Errorf("cache entry for %s names no namespace or interface", c.ContainerID)
+	}
+	for _, a := range c.Result.IPs {
+		ip, _, err := net.ParseCIDR(a.Address)
+		if err == nil && ip.To4() != nil {
+			return attachment{network: c.NetworkName, containerID: c.ContainerID, netns: c.NetNS, ifName: c.IfName, ip: ip.To4()}, nil
+		}
+	}
+	return attachment{}, fmt.Errorf("cache entry for %s has no IPv4 address", c.ContainerID)
+}

prog/weaveutil/portlock_test.go

@@ -0,0 +1,37 @@
+package main
+
+import (
+	"net"
+	"testing"
+)
+
+// shaped like CRI-O's libcni cache on the SG nodes (cniCacheV1, CNI 1.0.0 result); the config is elided
+const cacheEntry = `{"kind":"cniCacheV1","containerId":"2b61982132d6039b210e2fa5896555c8a921a07c756bd8753963440e07dc5f98",
+"config":"eyJ9","ifName":"eth0","networkName":"weave","netns":"/var/run/netns/8828bbba-9a68-460c-980b-266dc3c9342c",
+"cniArgs":[["K8S_POD_NAMESPACE","default"]],"capabilityArgs":{},
+"result":{"cniVersion":"1.0.0","ips":[{"address":"10.44.0.12/12","gateway":"10.44.0.0"}]}}`
+
+func TestParseAttachment(t *testing.T) {
+	a, err := parseAttachment([]byte(cacheEntry))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if a.network != "weave" || a.ifName != "eth0" || a.netns != "/var/run/netns/8828bbba-9a68-460c-980b-266dc3c9342c" ||
+		!a.ip.Equal(net.ParseIP("10.44.0.12")) || a.containerID[:7] != "2b61982" {
+		t.Fatalf("parsed %+v", a)
+	}
+}
+
+func TestParseAttachmentRefusesWhatItCannotLock(t *testing.T) {
+	for name, entry := range map[string]string{
+		"other kind":   `{"kind":"cniCacheV2","netns":"/x","ifName":"eth0","result":{"ips":[{"address":"10.0.0.1/8"}]}}`,
+		"no netns":     `{"kind":"cniCacheV1","ifName":"eth0","result":{"ips":[{"address":"10.0.0.1/8"}]}}`,
+		"no ipv4":      `{"kind":"cniCacheV1","netns":"/x","ifName":"eth0","result":{"ips":[{"address":"fd00::1/64"}]}}`,
+		"not json":     `{`,
+		"no addresses": `{"kind":"cniCacheV1","netns":"/x","ifName":"eth0","result":{}}`,
+	} {
+		if _, err := parseAttachment([]byte(entry)); err == nil {
+			t.Errorf("%s: want an error", name)
+		}
+	}
+}

reweave/test/egress/egress-test.yaml

@@ -0,0 +1,76 @@
+# End-to-end test of EgressService (weave-npc's lb agent). Placeholders are filled by run.sh: ADDR (an
+# address HOLDER holds), HOLDER, REMOTE and OTHER (the cluster's other two nodes).
+apiVersion: v1
+kind: List
+items:
+  - apiVersion: v1
+    kind: Namespace
+    metadata:
+      name: egtest
+  - apiVersion: v1
+    kind: Service
+    metadata:
+      name: mail
+      namespace: egtest
+    spec:
+      type: LoadBalancer
+      loadBalancerClass: reweave.levk.dev/node-address
+      loadBalancerIP: ADDR
+      allocateLoadBalancerNodePorts: false
+      selector: {egress: member}
+      ports: [{name: probe, port: 18090, targetPort: 8080}]
+  - apiVersion: reweave.levk.dev/v1alpha1
+    kind: EgressService
+    metadata:
+      name: mail
+      namespace: egtest
+    spec:
+      emit: Holder
+  # members on every node, and one that is not a member
+  - apiVersion: v1
+    kind: Pod
+    metadata: {name: member-remote, namespace: egtest, labels: {egress: member}}
+    spec:
+      nodeName: REMOTE
+      containers:
+        # root + NET_RAW: ping and traceroute need raw sockets (the image runs as an unprivileged user)
+        - {name: c, image: "curlimages/curl:8.10.1", command: ["sleep", "3600"], securityContext: {runAsUser: 0, capabilities: {add: [NET_RAW]}}}
+  - apiVersion: v1
+    kind: Pod
+    metadata: {name: member-other, namespace: egtest, labels: {egress: member}}
+    spec:
+      nodeName: OTHER
+      containers:
+        # root + NET_RAW: ping and traceroute need raw sockets (the image runs as an unprivileged user)
+        - {name: c, image: "curlimages/curl:8.10.1", command: ["sleep", "3600"], securityContext: {runAsUser: 0, capabilities: {add: [NET_RAW]}}}
+  - apiVersion: v1
+    kind: Pod
+    metadata: {name: member-holder, namespace: egtest, labels: {egress: member}}
+    spec:
+      nodeName: HOLDER
+      containers:
+        # root + NET_RAW: ping and traceroute need raw sockets (the image runs as an unprivileged user)
+        - {name: c, image: "curlimages/curl:8.10.1", command: ["sleep", "3600"], securityContext: {runAsUser: 0, capabilities: {add: [NET_RAW]}}}
+  - apiVersion: v1
+    kind: Pod
+    metadata: {name: member-dns, namespace: egtest, labels: {egress: member}}
+    spec:
+      nodeName: REMOTE
+      containers: [{name: c, image: "registry.k8s.io/e2e-test-images/jessie-dnsutils:1.7", command: ["sleep", "3600"]}]
+  - apiVersion: v1
+    kind: Pod
+    metadata: {name: plain, namespace: egtest, labels: {egress: "none"}}
+    spec:
+      nodeName: REMOTE
+      containers: [{name: c, image: "curlimages/curl:8.10.1", command: ["sleep", "3600"]}]
+  # east-west: /clientip answers with the address the connection came from
+  - apiVersion: v1
+    kind: Pod
+    metadata: {name: echo, namespace: egtest, labels: {app: echo}}
+    spec:
+      nodeName: OTHER
+      containers:
+        - name: echo
+          image: registry.k8s.io/e2e-test-images/agnhost:2.53
+          args: ["netexec", "--http-port=8080"]
+          readinessProbe: {httpGet: {path: /healthz, port: 8080}}

reweave/test/egress/run.sh

@@ -0,0 +1,120 @@
+#!/usr/bin/env bash
+# End-to-end test of EgressService against a live cluster (assertions A1-A6 of the design, plus the first
+# packet). Members ask outside services which address they come from: api.ipify.org (TCP) and OpenDNS's
+# myip.opendns.com (UDP).
+#
+#   CP="ssh ... root@<cp>" HOLDER_SSH="ssh ... root@<holder>" REMOTE_SSH="ssh ... root@<remote>" \
+#   ADDR=<address HOLDER holds> HOLDER=<node> REMOTE=<node> OTHER=<node> \
+#   reweave/test/egress/run.sh
+#
+# Asserts, in order:
+#   0. the holder writes the status: host, address, a port block for every node
+#   1. the first packet of a member started after the EgressService leaves as ADDR (CNI admission)
+#   A1/A2. members on every node leave as ADDR (TCP and UDP) and get their replies; ping works from every node
+#   ICMP errors from routers on the way (traceroute's time-exceeded) reach a member on another node
+#   A4. a non-member pod on the same node as a member leaves as its node (untouched), and no rule names it
+#   A3. a member talking to a pod sees its own pod address there (east-west untouched)
+#   A5. naming an address nobody holds clears the status, and members' outside traffic (TCP, ping) is dropped
+#   A6. deleting the EgressService removes its state: members leave as their node again, rules are gone
+# Leaves nothing behind (namespace egtest is deleted, also on failure).
+set -euo pipefail
+
+: "${CP:?}" "${HOLDER_SSH:?}" "${REMOTE_SSH:?}" "${ADDR:?}" "${HOLDER:?}" "${REMOTE:?}" "${OTHER:?}"
+here=$(cd "$(dirname "$0")" && pwd)
+k() { $CP "KUBECONFIG=/etc/kubernetes/admin.conf kubectl $*"; }
+fail() { echo "FAIL: $*" >&2; exit 1; }
+ok() { echo "ok: $*"; }
+cleanup() { k delete namespace egtest --ignore-not-found --wait=false >/dev/null 2>&1 || true; }
+trap cleanup EXIT
+
+eventually() {
+    local n=$1; shift
+    for _ in $(seq "$n"); do
+        if "$@" >/dev/null 2>&1; then return 0; fi
+        sleep 1
+    done
+    return 1
+}
+whoami_tcp() { k -n egtest exec "$1" -- curl -s --max-time 10 https://api.ipify.org; }
+# each check re-reads on every try of eventually
+tcp_is() { [ "$(whoami_tcp "$1" || true)" = "$2" ]; }
+whoami_udp() { k -n egtest exec "$1" -- dig +short +time=5 +tries=1 myip.opendns.com @208.67.222.222; }
+node_ip() { k get node "$1" -o "jsonpath='{.status.addresses[?(@.type==\"InternalIP\")].address}'"; }
+
+remote_ip=$(node_ip "$REMOTE")
+[ -n "$remote_ip" ] || fail "no address for $REMOTE"
+
+sed -e "s/\bADDR\b/$ADDR/g; s/\bHOLDER\b/$HOLDER/g; s/\bREMOTE\b/$REMOTE/g; s/\bOTHER\b/$OTHER/g" \
+    "$here/egress-test.yaml" | $CP "KUBECONFIG=/etc/kubernetes/admin.conf kubectl apply -f -" >/dev/null
+k -n egtest wait --for=condition=Ready pod --all --timeout=300s >/dev/null || fail "test pods not ready"
+
+status() { k -n egtest get egressservice mail -o "jsonpath='{.status.host} {.status.address} {.status.blocks[*].node}'"; }
+status_is() { [ "$(status)" = "$1" ]; }
+first_line() { k -n egtest logs member-first 2>/dev/null | head -1; }
+said_something() { [ -n "$(first_line)" ]; }
+want="$HOLDER $ADDR $(printf '%s\n' "$HOLDER" "$REMOTE" "$OTHER" | sort | tr '\n' ' ' | sed 's/ $//')"
+eventually 60 status_is "$want" || fail "status: '$(status)', want '$want'"
+ok "0. status: host $HOLDER, address $ADDR, a port block for each node"
+
+# a member created now: its very first connection must already leave as ADDR
+$CP "KUBECONFIG=/etc/kubernetes/admin.conf kubectl -n egtest run member-first --labels egress=member --restart=Never \
+    --overrides='{\"spec\":{\"nodeName\":\"$REMOTE\"}}' --image=curlimages/curl:8.10.1 --command -- \
+    sh -c 'curl -s --max-time 10 https://api.ipify.org; echo; sleep 3600'" >/dev/null
+eventually 120 said_something || fail "member-first said nothing"
+first=$(first_line)
+[ "$first" = "$ADDR" ] || fail "member-first's first connection left as '$first', want $ADDR"
+ok "1. a new member's first connection leaves as $ADDR"
+
+for p in member-remote member-other member-holder; do
+    got=$(whoami_tcp $p || true)
+    [ "$got" = "$ADDR" ] || fail "A1: $p left as '$got', want $ADDR"
+done
+got=$(whoami_udp member-dns || true)
+[ "$got" = "$ADDR" ] || fail "A1: member-dns (UDP) left as '$got', want $ADDR"
+ok "A1/A2. members on $REMOTE, $OTHER and $HOLDER leave as $ADDR over TCP and UDP, replies come back"
+
+for p in member-remote member-other member-holder; do
+    k -n egtest exec $p -- ping -c 2 -W 3 1.1.1.1 >/dev/null || fail "ping from $p got no reply"
+done
+ok "ping from members on all three nodes gets its replies"
+
+# hop 1 is the member's own node, hop 2 the holder (it answers its own address, the member cannot see it); every
+# later hop is a router on the internet whose time-exceeded comes back to ADDR and must find the member
+hops=$(k -n egtest exec member-remote -- traceroute -n -m 8 -q 1 -w 2 1.1.1.1 2>&1 || true)
+echo "$hops" | awk '$1 >= 3 && $2 ~ /^[0-9]+\./ {found=1} END {exit !found}' \
+    || fail "no router beyond the holder answered the member's traceroute:
+$hops"
+ok "ICMP errors from routers beyond the holder reach a member on $REMOTE (traceroute)"
+
+got=$(whoami_tcp plain || true)
+[ "$got" = "$remote_ip" ] || fail "A4: non-member left as '$got', want its node $remote_ip"
+plain_ip=$(k -n egtest get pod plain -o "jsonpath='{.status.podIP}'")
+if $REMOTE_SSH "iptables-save | grep -F -- '$plain_ip/32'; ip rule | grep -F -- '$plain_ip'"; then
+    fail "A4: a rule names the non-member $plain_ip"
+fi
+ok "A4. the non-member leaves as its node ($remote_ip); no rule names it"
+
+echo_ip=$(k -n egtest get pod echo -o "jsonpath='{.status.podIP}'")
+member_ip=$(k -n egtest get pod member-remote -o "jsonpath='{.status.podIP}'")
+got=$(k -n egtest exec member-remote -- curl -s --max-time 5 "http://$echo_ip:8080/clientip" || true)
+[ "${got%:*}" = "$member_ip" ] || fail "A3: echo saw '$got', want the member's pod address $member_ip"
+ok "A3. east-west: the echo pod sees the member's own address ($member_ip)"
+
+k -n egtest patch svc mail --type merge -p "'{\"spec\":{\"loadBalancerIP\":\"192.0.2.1\"}}'" >/dev/null
+eventually 60 status_is "  " || fail "A5: status not cleared: '$(status)'"
+got=$(whoami_tcp member-remote || true)
+[ -z "$got" ] || fail "A5: with no holder a member still got out, as '$got'"
+if k -n egtest exec member-remote -- ping -c 1 -W 3 1.1.1.1 >/dev/null 2>&1; then
+    fail "A5: with no holder a member's ping still got out"
+fi
+k -n egtest patch svc mail --type merge -p "'{\"spec\":{\"loadBalancerIP\":\"$ADDR\"}}'" >/dev/null
+eventually 60 status_is "$want" || fail "status not restored: '$(status)'"
+eventually 30 tcp_is member-remote "$ADDR" || fail "member did not get back out as $ADDR"
+ok "A5. no holder: status cleared, members dropped (not leaked); restored afterwards"
+
+k -n egtest delete egressservice mail >/dev/null
+eventually 60 tcp_is member-remote "$remote_ip" || fail "A6: after delete, member left as '$(whoami_tcp member-remote || true)'"
+if $REMOTE_SSH "ip rule | grep -E '^9100:'"; then fail "A6: member rules left on $REMOTE"; fi
+if $HOLDER_SSH "ip rule | grep -E '^90:'"; then fail "A6: holder rules left on $HOLDER"; fi
+ok "A6. EgressService deleted: members leave as their node again, no rules left"
+echo PASS

reweave/test/lb/run.sh

@@ -42,9 +42,12 @@ sed -e "s/\bADDR\b/$ADDR/g; s/\bHOLDER\b/$HOLDER/g; s/\bREMOTE\b/$REMOTE/g; s/\b
 k -n lbtest wait --for=condition=Ready pod --all --timeout=180s >/dev/null || fail "test pods not ready"
 
 status() { k -n lbtest get svc "$1" -o "jsonpath='{.status.loadBalancer.ingress}'"; }
+# each check re-reads on every try of eventually
+status_is() { [ "$(status "$1")" = "$2" ]; }
+chains_gone() { [ "$($HOLDER_SSH "iptables-save -t nat | grep -c '^:REWEAVE-LBS-'" || true)" = 0 ]; }
 for s in echo-remote echo-local; do
     want="[{\"ip\":\"$ADDR\",\"ipMode\":\"Proxy\"}]"
-    eventually 30 test "$(status $s)" = "$want" || fail "$s status: $(status $s), want $want"
+    eventually 30 status_is $s "$want" || fail "$s status: $(status $s), want $want"
 done
 ok "1. status [{ip: $ADDR, ipMode: Proxy}] on both Services"
 
@@ -70,14 +73,13 @@ k -n lbtest exec client -- /agnhost connect "$ADDR:18080" --timeout 5s >/dev/nul
 ok "4. a pod on $OTHER reaches $ADDR:18080"
 
 k -n lbtest patch svc echo-remote --type merge -p "'{\"spec\":{\"loadBalancerIP\":\"192.0.2.1\"}}'" >/dev/null
-eventually 30 test "$(status echo-remote)" = "" || fail "status not cleared for an unheld address: $(status echo-remote)"
+eventually 30 status_is echo-remote "" || fail "status not cleared for an unheld address: $(status echo-remote)"
 if curl -s --max-time 3 "http://$ADDR:18080/clientip" >/dev/null; then
     fail "$ADDR:18080 still forwards after its Service moved to an unheld address"
 fi
 ok "5. unheld address: status cleared, forward gone (fail closed)"
 
 k -n lbtest delete svc echo-remote echo-local >/dev/null
-eventually 30 test "$($HOLDER_SSH "iptables-save -t nat | grep -c '^:REWEAVE-LBS-'" || true)" = 0 \
-    || fail "holder kept per-port chains after the Services went away"
+eventually 30 chains_gone || fail "holder kept per-port chains after the Services went away"
 ok "6. Services deleted: holder's chains removed"
 echo PASS

reweave/test/portlock/netns.py

@@ -0,0 +1,14 @@
+# netns.py <namespace> <pod>: the network namespace the weave CNI plugin attached a pod in, from the runtime's
+# record of the attachment (libcni's result cache). Runs on the node.
+import glob
+import json
+import sys
+
+namespace, pod = sys.argv[1], sys.argv[2]
+for path in glob.glob("/var/lib/cni/results/weave-*"):
+    with open(path) as f:
+        entry = json.load(f)
+    args = dict(entry.get("cniArgs") or [])
+    if args.get("K8S_POD_NAMESPACE") == namespace and args.get("K8S_POD_NAME") == pod:
+        print(entry["netns"])
+        break

reweave/test/portlock/portlock-test.yaml

@@ -0,0 +1,30 @@
+# Pods for the port lock e2e (run.sh): a forger and a victim on node A, a victim on node B. None has any added
+# capability; the forged frames are sent from the forger's network namespace by root on the node.
+apiVersion: v1
+kind: Namespace
+metadata:
+  name: pltest
+---
+apiVersion: v1
+kind: Pod
+metadata: {name: forger, namespace: pltest}
+spec:
+  nodeName: NODE_A
+  containers:
+    - {name: c, image: "curlimages/curl:8.10.1", command: ["sleep", "3600"]}
+---
+apiVersion: v1
+kind: Pod
+metadata: {name: victim-local, namespace: pltest}
+spec:
+  nodeName: NODE_A
+  containers:
+    - {name: c, image: "curlimages/curl:8.10.1", command: ["sleep", "3600"]}
+---
+apiVersion: v1
+kind: Pod
+metadata: {name: victim-remote, namespace: pltest}
+spec:
+  nodeName: NODE_B
+  containers:
+    - {name: c, image: "curlimages/curl:8.10.1", command: ["sleep", "3600"]}

reweave/test/portlock/probe/main.go

@@ -0,0 +1,200 @@
+// Command probe sends hand-made Ethernet frames out of an interface, or waits for one carrying a tag. The port lock
+// e2e runs it inside pods' network namespaces (nsenter, as root on the node): a pod forging another's addresses
+// must not reach anyone.
+//
+//	probe send <ifname> <dst-mac> <src-mac|own> ipv4|arp <src-ip|own> <dst-ip> <tag>
+//	probe listen <ifname> <tag> <seconds>
+package main
+
+import (
+	"bytes"
+	"encoding/binary"
+	"fmt"
+	"net"
+	"os"
+	"strconv"
+	"time"
+
+	"golang.org/x/sys/unix"
+)
+
+func main() {
+	if len(os.Args) < 2 {
+		usage()
+	}
+	var err error
+	switch os.Args[1] {
+	case "send":
+		if len(os.Args) != 9 {
+			usage()
+		}
+		err = send(os.Args[2], os.Args[3], os.Args[4], os.Args[5], os.Args[6], os.Args[7], os.Args[8])
+	case "listen":
+		if len(os.Args) != 5 {
+			usage()
+		}
+		secs, _ := strconv.Atoi(os.Args[4])
+		var seen bool
+		seen, err = listen(os.Args[2], os.Args[3], time.Duration(secs)*time.Second)
+		if err == nil {
+			if seen {
+				fmt.Println("seen")
+				return
+			}
+			fmt.Println("not seen")
+			os.Exit(1)
+		}
+	default:
+		usage()
+	}
+	if err != nil {
+		fmt.Fprintln(os.Stderr, err)
+		os.Exit(2)
+	}
+}
+
+func usage() {
+	fmt.Fprintln(os.Stderr, "usage: probe send <ifname> <dst-mac> <src-mac|own> ipv4|arp <src-ip|own> <dst-ip> <tag>")
+	fmt.Fprintln(os.Stderr, "       probe listen <ifname> <tag> <seconds>")
+	os.Exit(2)
+}
+
+func send(ifName, dstMAC, srcMAC, kind, srcIP, dstIP, tag string) error {
+	iface, err := net.InterfaceByName(ifName)
+	if err != nil {
+		return err
+	}
+	dst, err := net.ParseMAC(dstMAC)
+	if err != nil {
+		return err
+	}
+	src := iface.HardwareAddr
+	if srcMAC != "own" {
+		if src, err = net.ParseMAC(srcMAC); err != nil {
+			return err
+		}
+	}
+	var from net.IP
+	if srcIP == "own" {
+		addrs, err := iface.Addrs()
+		if err != nil {
+			return err
+		}
+		for _, a := range addrs {
+			if n, ok := a.(*net.IPNet); ok && n.IP.To4() != nil {
+				from = n.IP.To4()
+			}
+		}
+		if from == nil {
+			return fmt.Errorf("%s has no IPv4 address", ifName)
+		}
+	} else if from = net.ParseIP(srcIP).To4(); from == nil {
+		return fmt.Errorf("bad source address %q", srcIP)
+	}
+	to := net.ParseIP(dstIP).To4()
+	if to == nil {
+		return fmt.Errorf("bad destination address %q", dstIP)
+	}
+	var frame []byte
+	switch kind {
+	case "ipv4":
+		frame = ipv4Frame(dst, src, from, to, []byte(tag))
+	case "arp":
+		frame = arpFrame(dst, src, from, to, []byte(tag))
+	default:
+		return fmt.Errorf("unknown kind %q", kind)
+	}
+	fd, err := unix.Socket(unix.AF_PACKET, unix.SOCK_RAW, int(htons(unix.ETH_P_ALL)))
+	if err != nil {
+		return err
+	}
+	defer unix.Close(fd)
+	if err := unix.Bind(fd, &unix.SockaddrLinklayer{Protocol: htons(unix.ETH_P_ALL), Ifindex: iface.Index}); err != nil {
+		return err
+	}
+	// a few copies, spaced out: the bridge and weave may still be learning where the destination is
+	for i := 0; i < 3; i++ {
+		if _, err := unix.Write(fd, frame); err != nil {
+			return err
+		}
+		time.Sleep(200 * time.Millisecond)
+	}
+	return nil
+}
+
+func listen(ifName, tag string, within time.Duration) (bool, error) {
+	iface, err := net.InterfaceByName(ifName)
+	if err != nil {
+		return false, err
+	}
+	fd, err := unix.Socket(unix.AF_PACKET, unix.SOCK_RAW, int(htons(unix.ETH_P_ALL)))
+	if err != nil {
+		return false, err
+	}
+	defer unix.Close(fd)
+	if err := unix.Bind(fd, &unix.SockaddrLinklayer{Protocol: htons(unix.ETH_P_ALL), Ifindex: iface.Index}); err != nil {
+		return false, err
+	}
+	tv := unix.NsecToTimeval((100 * time.Millisecond).Nanoseconds())
+	if err := unix.SetsockoptTimeval(fd, unix.SOL_SOCKET, unix.SO_RCVTIMEO, &tv); err != nil {
+		return false, err
+	}
+	buf := make([]byte, 65536)
+	deadline := time.Now().Add(within)
+	for time.Now().Before(deadline) {
+		n, _, err := unix.Recvfrom(fd, buf, 0)
+		if err != nil {
+			continue
+		}
+		if bytes.Contains(buf[:n], []byte(tag)) {
+			return true, nil
+		}
+	}
+	return false, nil
+}
+
+func htons(v uint16) uint16 { return v<<8 | v>>8 }
+
+func ether(dst, src net.HardwareAddr, ethertype uint16, payload []byte) []byte {
+	f := make([]byte, 14, 14+len(payload))
+	copy(f[0:6], dst)
+	copy(f[6:12], src)
+	binary.BigEndian.PutUint16(f[12:14], ethertype)
+	return append(f, payload...)
+}
+
+func ipv4Frame(dst, src net.HardwareAddr, from, to net.IP, body []byte) []byte {
+	h := make([]byte, 20)
+	h[0] = 0x45
+	binary.BigEndian.PutUint16(h[2:4], uint16(20+len(body)))
+	h[8] = 64
+	h[9] = 253 // experimental protocol: nothing answers it, nothing on the way acts on it
+	copy(h[12:16], from)
+	copy(h[16:20], to)
+	binary.BigEndian.PutUint16(h[10:12], checksum(h))
+	return ether(dst, src, unix.ETH_P_IP, append(h, body...))
+}
+
+// arpFrame is an ARP request speaking for (src, from), asking for to; the tag rides as Ethernet trailer.
+func arpFrame(dst, src net.HardwareAddr, from, to net.IP, tag []byte) []byte {
+	a := make([]byte, 28)
+	binary.BigEndian.PutUint16(a[0:2], 1)
+	binary.BigEndian.PutUint16(a[2:4], unix.ETH_P_IP)
+	a[4], a[5] = 6, 4
+	binary.BigEndian.PutUint16(a[6:8], 1)
+	copy(a[8:14], src)
+	copy(a[14:18], from)
+	copy(a[24:28], to)
+	return ether(dst, src, unix.ETH_P_ARP, append(a, tag...))
+}
+
+func checksum(b []byte) uint16 {
+	var s uint32
+	for i := 0; i+1 < len(b); i += 2 {
+		s += uint32(binary.BigEndian.Uint16(b[i:]))
+	}
+	for s>>16 != 0 {
+		s = s&0xffff + s>>16
+	}
+	return ^uint16(s)
+}

reweave/test/portlock/run.sh

@@ -0,0 +1,112 @@
+#!/usr/bin/env bash
+# End-to-end test of the port lock against a live cluster (assertion A7): a pod speaks only as itself.
+#
+#   CP="ssh ... root@<cp>" A_SSH="ssh ... root@<a>" B_SSH="ssh ... root@<b>" A=<node a> B=<node b> \
+#   reweave/test/portlock/run.sh
+#
+# Asserts, in order:
+#   0. every pod port on the three nodes (the CP's, A's and B's) carries the lock
+#   1. a pod's own frames reach a pod on its node and a pod on another node (the control)
+#   2. frames forging another pod's address reach neither
+#   3. frames forging a node's bridge (its weave MAC and address) reach neither
+#   4. ARP claiming another pod's address reaches no one; ARP for itself does
+# The forged frames are sent from the forging pod's network namespace by root on its node (nsenter), the most a
+# pod could ever do; the pod itself has no added capability.
+# Leaves nothing behind (namespace pltest is deleted, also on failure; the probe is removed from the nodes).
+set -euo pipefail
+
+: "${CP:?}" "${A_SSH:?}" "${B_SSH:?}" "${A:?}" "${B:?}"
+here=$(cd "$(dirname "$0")" && pwd)
+k() { $CP "KUBECONFIG=/etc/kubernetes/admin.conf kubectl $*"; }
+fail() { echo "FAIL: $*" >&2; exit 1; }
+ok() { echo "ok: $*"; }
+work=$(mktemp -d)
+cleanup() {
+    k delete namespace pltest --ignore-not-found --wait=false >/dev/null 2>&1 || true
+    for s in "$A_SSH" "$B_SSH"; do $s 'rm -f /root/plprobe /root/plnetns.py' >/dev/null 2>&1 || true; done
+    rm -rf "$work"
+}
+trap cleanup EXIT
+
+(cd "$here/../../.." && CGO_ENABLED=0 go build -o "$work/probe" ./reweave/test/portlock/probe/)
+for s in "$A_SSH" "$B_SSH"; do
+    $s 'cat > /root/plprobe && chmod +x /root/plprobe' < "$work/probe"
+    $s 'cat > /root/plnetns.py' < "$here/netns.py"
+done
+
+# a port is locked when its ingress carries the three passes and the drop
+unlocked() {
+    $1 'for p in $(ip -br link show master weave | sed "s/@.*//; s/ .*//" | grep ^vethwepl); do
+            f=$(tc filter show dev $p ingress)
+            for want in "protocol ip pref 1 u32" "protocol arp pref 2 u32" "protocol arp pref 3 u32" \
+                        "protocol all pref 9 matchall" "gact action drop"; do
+                echo "$f" | grep -q "$want" || { echo "$p"; break; }
+            done
+        done'
+}
+all_locked() {
+    for s in "$CP" "$A_SSH" "$B_SSH"; do
+        u=$(unlocked "$s")
+        [ -z "$u" ] || fail "0. unlocked pod ports ($1): $u"
+    done
+}
+all_locked "before the test pods"
+ok "0. every pod port on the three nodes carries the lock"
+
+sed -e "s/\bNODE_A\b/$A/g; s/\bNODE_B\b/$B/g" "$here/portlock-test.yaml" |
+    $CP "KUBECONFIG=/etc/kubernetes/admin.conf kubectl apply -f -" >/dev/null
+k -n pltest wait --for=condition=Ready pod --all --timeout=300s >/dev/null || fail "test pods not ready"
+all_locked "with the test pods"
+
+netns_of() { $1 "python3 /root/plnetns.py pltest $2"; }
+mac_in() { $1 "nsenter --net=$2 ip -br link show eth0" | awk '{print $3}'; }
+pod_ip() { k -n pltest get pod "$1" -o "jsonpath='{.status.podIP}'"; }
+f_ns=$(netns_of "$A_SSH" forger)
+vl_ns=$(netns_of "$A_SSH" victim-local)
+vr_ns=$(netns_of "$B_SSH" victim-remote)
+[ -n "$f_ns" ] && [ -n "$vl_ns" ] && [ -n "$vr_ns" ] || fail "test pods' namespaces not found"
+vl_mac=$(mac_in "$A_SSH" "$vl_ns")
+vr_mac=$(mac_in "$B_SSH" "$vr_ns")
+vl_ip=$(pod_ip victim-local)
+vr_ip=$(pod_ip victim-remote)
+# the nodes' bridges, from the annotation the lb agent writes ("ip,mac")
+bridge_of() { k get node "$1" -o "jsonpath='{.metadata.annotations.reweave\.levk\.dev/weave}'"; }
+b_bridge=$(bridge_of "$B"); b_ip=${b_bridge%,*}; b_mac=${b_bridge#*,}
+a_bridge=$(bridge_of "$A"); a_ip=${a_bridge%,*}; a_mac=${a_bridge#*,}
+[ -n "$b_ip" ] && [ -n "$b_mac" ] && [ -n "$a_ip" ] && [ -n "$a_mac" ] || fail "no weave bridge annotation on $A or $B"
+
+# expect <seen|not seen> <victim: local|remote> <tag> <send args...>: the victim listens while the forger sends
+expect() {
+    local want=$1 where=$2 tag=$3-$RANDOM
+    shift 3
+    local vs vns
+    if [ "$where" = local ]; then vs=$A_SSH vns=$vl_ns; else vs=$B_SSH vns=$vr_ns; fi
+    $vs "nsenter --net=$vns /root/plprobe listen eth0 $tag 6" > "$work/listen" 2>&1 &
+    local listener=$!
+    sleep 1
+    $A_SSH "nsenter --net=$f_ns /root/plprobe send eth0 $* $tag" || fail "probe could not send: $*"
+    wait $listener || true
+    local got
+    got=$(cat "$work/listen")
+    [ "$got" = "$want" ] || fail "$tag ($where): '$got', want '$want'"
+}
+
+expect "seen" local own-local "$vl_mac own ipv4 own $vl_ip"
+expect "seen" remote own-remote "$vr_mac own ipv4 own $vr_ip"
+ok "1. the forger's own frames reach a pod on its node and one on $B"
+
+expect "not seen" remote pod-ip-remote "$vr_mac own ipv4 $vl_ip $vr_ip"
+expect "not seen" local pod-ip-local "$vl_mac own ipv4 $vr_ip $vl_ip"
+expect "not seen" remote pod-mac-remote "$vr_mac $vl_mac ipv4 $vl_ip $vr_ip"
+ok "2. frames forging another pod's address reach no one"
+
+expect "not seen" local node-b-local "$vl_mac $b_mac ipv4 $b_ip $vl_ip"
+expect "not seen" remote node-b-remote "$vr_mac $b_mac ipv4 $b_ip $vr_ip"
+expect "not seen" remote node-a-remote "$vr_mac $a_mac ipv4 $a_ip $vr_ip"
+ok "3. frames forging a node's bridge ($A's, $B's) reach no one"
+
+expect "not seen" local arp-claim-local "ff:ff:ff:ff:ff:ff own arp $vr_ip $vl_ip"
+expect "not seen" remote arp-claim-remote "ff:ff:ff:ff:ff:ff own arp $vl_ip $vr_ip"
+expect "seen" remote arp-own-remote "ff:ff:ff:ff:ff:ff own arp own $vr_ip"
+ok "4. ARP claiming another pod's address reaches no one; the forger's ARP for itself does"
+echo "PASS"

weave

@@ -725,8 +725,11 @@ peer_args() {
 
 install_cni_plugin() {
     mkdir -p $1 || return 1
-    if [ ! -f "$1/$CNI_PLUGIN_NAME" ]; then
-        cp /usr/bin/weaveutil "$1/$CNI_PLUGIN_NAME"
+    # (re)install whenever the image's plugin differs from the installed one: the name
+    # (weave-plugin-$IMAGE_VERSION) stays the same across builds of one version ("latest")
+    if [ ! -f "$1/$CNI_PLUGIN_NAME" ] || \
+       [ "$(sha256sum < /usr/bin/weaveutil)" != "$(sha256sum < "$1/$CNI_PLUGIN_NAME")" ]; then
+        cp /usr/bin/weaveutil "$1/$CNI_PLUGIN_NAME.tmp" && mv -f "$1/$CNI_PLUGIN_NAME.tmp" "$1/$CNI_PLUGIN_NAME"
     fi
 }
 
@@ -763,7 +766,8 @@ create_cni_config() {
         {
             "name": "weave",
             "type": "weave-net",
-            "hairpinMode": $2
+            "hairpinMode": $2,
+            "egressAdmission": $3
         },
         {
             "type": "portmap",
@@ -778,6 +782,9 @@ EOF
 setup_cni() {
     # if env var HAIRPIN_MODE is not set, default it to true
     HAIRPIN_MODE=${HAIRPIN_MODE:-true}
+    # weave-npc admits every pod before it gets its interface (EgressService members leave from their
+    # Service's address from the first packet); set to false only where weave-npc runs with --lb=false
+    EGRESS_ADMISSION=${EGRESS_ADMISSION:-true}
     if install_cni_plugin $CNI_PLUGIN_DIR ; then
         upgrade_cni_plugin $CNI_PLUGIN_DIR
     fi
@@ -786,8 +793,15 @@ setup_cni() {
     # clusters with Weave version <= 2.4.0 remove existing 10-weave.conf
     rm -f $HOST_ROOT/etc/cni/net.d/10-weave.conf
 
-    if [ -d $HOST_ROOT/etc/cni/net.d -a ! -f $HOST_ROOT/etc/cni/net.d/10-weave.conflist ] ; then
-        create_cni_config $HOST_ROOT/etc/cni/net.d/10-weave.conflist $HAIRPIN_MODE
+    # (re)write the conflist whenever what it should say changed, so upgrades reach existing nodes
+    if [ -d $HOST_ROOT/etc/cni/net.d ] ; then
+        CONFLIST=$HOST_ROOT/etc/cni/net.d/10-weave.conflist
+        create_cni_config $CONFLIST.new $HAIRPIN_MODE $EGRESS_ADMISSION
+        if [ -f $CONFLIST ] && [ "$(cat $CONFLIST.new)" = "$(cat $CONFLIST)" ] ; then
+            rm -f $CONFLIST.new
+        else
+            mv -f $CONFLIST.new $CONFLIST
+        fi
     fi
 }