weave/lb/egress_datapath.go

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)
		}
		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)
}