feat(rule): support rule subscribe

Author: eustrainLee

rule.go

@@ -46,6 +46,12 @@ func (r Rule) String() string {
 		r.Priority, from, to, r.Table, r.typeString())
 }
 
+// RuleUpdate is sent when a route changes - type is RTM_NEWRULE or RTM_DELRULE
+type RuleUpdate struct {
+	Type uint16
+	Rule
+}
+
 // NewRule return empty rules.
 func NewRule() *Rule {
 	return &Rule{

rule_linux.go

@@ -5,8 +5,10 @@ import (
 	"errors"
 	"fmt"
 	"net"
+	"syscall"
 
 	"github.com/vishvananda/netlink/nl"
+	"github.com/vishvananda/netns"
 	"golang.org/x/sys/unix"
 )
 
@@ -227,73 +229,11 @@ func (h *Handle) RuleListFiltered(family int, filter *Rule, filterMask uint64) (
 
 	var res = make([]Rule, 0)
 	for i := range msgs {
-		msg := nl.DeserializeRtMsg(msgs[i])
-		attrs, err := nl.ParseRouteAttr(msgs[i][msg.Len():])
+		rule, err := deserializeRule(msgs[i])
 		if err != nil {
 			return nil, err
 		}
 
-		rule := NewRule()
-		rule.Priority = 0 // The default priority from kernel
-
-		rule.Invert = msg.Flags&FibRuleInvert > 0
-		rule.Family = int(msg.Family)
-		rule.Tos = uint(msg.Tos)
-
-		for j := range attrs {
-			switch attrs[j].Attr.Type {
-			case unix.RTA_TABLE:
-				rule.Table = int(native.Uint32(attrs[j].Value[0:4]))
-			case nl.FRA_SRC:
-				rule.Src = &net.IPNet{
-					IP:   attrs[j].Value,
-					Mask: net.CIDRMask(int(msg.Src_len), 8*len(attrs[j].Value)),
-				}
-			case nl.FRA_DST:
-				rule.Dst = &net.IPNet{
-					IP:   attrs[j].Value,
-					Mask: net.CIDRMask(int(msg.Dst_len), 8*len(attrs[j].Value)),
-				}
-			case nl.FRA_FWMARK:
-				rule.Mark = native.Uint32(attrs[j].Value[0:4])
-			case nl.FRA_FWMASK:
-				mask := native.Uint32(attrs[j].Value[0:4])
-				rule.Mask = &mask
-			case nl.FRA_TUN_ID:
-				rule.TunID = uint(native.Uint64(attrs[j].Value[0:8]))
-			case nl.FRA_IIFNAME:
-				rule.IifName = string(attrs[j].Value[:len(attrs[j].Value)-1])
-			case nl.FRA_OIFNAME:
-				rule.OifName = string(attrs[j].Value[:len(attrs[j].Value)-1])
-			case nl.FRA_SUPPRESS_PREFIXLEN:
-				i := native.Uint32(attrs[j].Value[0:4])
-				if i != 0xffffffff {
-					rule.SuppressPrefixlen = int(i)
-				}
-			case nl.FRA_SUPPRESS_IFGROUP:
-				i := native.Uint32(attrs[j].Value[0:4])
-				if i != 0xffffffff {
-					rule.SuppressIfgroup = int(i)
-				}
-			case nl.FRA_FLOW:
-				rule.Flow = int(native.Uint32(attrs[j].Value[0:4]))
-			case nl.FRA_GOTO:
-				rule.Goto = int(native.Uint32(attrs[j].Value[0:4]))
-			case nl.FRA_PRIORITY:
-				rule.Priority = int(native.Uint32(attrs[j].Value[0:4]))
-			case nl.FRA_IP_PROTO:
-				rule.IPProto = int(native.Uint32(attrs[j].Value[0:4]))
-			case nl.FRA_DPORT_RANGE:
-				rule.Dport = NewRulePortRange(native.Uint16(attrs[j].Value[0:2]), native.Uint16(attrs[j].Value[2:4]))
-			case nl.FRA_SPORT_RANGE:
-				rule.Sport = NewRulePortRange(native.Uint16(attrs[j].Value[0:2]), native.Uint16(attrs[j].Value[2:4]))
-			case nl.FRA_UID_RANGE:
-				rule.UIDRange = NewRuleUIDRange(native.Uint32(attrs[j].Value[0:4]), native.Uint32(attrs[j].Value[4:8]))
-			case nl.FRA_PROTOCOL:
-				rule.Protocol = uint8(attrs[j].Value[0])
-			}
-		}
-
 		if filter != nil {
 			switch {
 			case filterMask&RT_FILTER_SRC != 0 &&
@@ -376,3 +316,193 @@ func (r Rule) typeString() string {
 		return fmt.Sprintf("type(0x%x)", r.Type)
 	}
 }
+
+// deserializeRule decodes a binary netlink message into a Rule struct
+func deserializeRule(m []byte) (*Rule, error) {
+	msg := nl.DeserializeRtMsg(m)
+	attrs, err := nl.ParseRouteAttr(m[msg.Len():])
+	if err != nil {
+		return nil, err
+	}
+
+	rule := NewRule()
+	rule.Priority = 0 // The default priority from kernel
+
+	rule.Invert = msg.Flags&FibRuleInvert > 0
+	rule.Family = int(msg.Family)
+	rule.Tos = uint(msg.Tos)
+
+	for j := range attrs {
+		switch attrs[j].Attr.Type {
+		case unix.RTA_TABLE:
+			rule.Table = int(native.Uint32(attrs[j].Value[0:4]))
+		case nl.FRA_SRC:
+			rule.Src = &net.IPNet{
+				IP:   attrs[j].Value,
+				Mask: net.CIDRMask(int(msg.Src_len), 8*len(attrs[j].Value)),
+			}
+		case nl.FRA_DST:
+			rule.Dst = &net.IPNet{
+				IP:   attrs[j].Value,
+				Mask: net.CIDRMask(int(msg.Dst_len), 8*len(attrs[j].Value)),
+			}
+		case nl.FRA_FWMARK:
+			rule.Mark = native.Uint32(attrs[j].Value[0:4])
+		case nl.FRA_FWMASK:
+			mask := native.Uint32(attrs[j].Value[0:4])
+			rule.Mask = &mask
+		case nl.FRA_TUN_ID:
+			rule.TunID = uint(native.Uint64(attrs[j].Value[0:8]))
+		case nl.FRA_IIFNAME:
+			rule.IifName = string(attrs[j].Value[:len(attrs[j].Value)-1])
+		case nl.FRA_OIFNAME:
+			rule.OifName = string(attrs[j].Value[:len(attrs[j].Value)-1])
+		case nl.FRA_SUPPRESS_PREFIXLEN:
+			i := native.Uint32(attrs[j].Value[0:4])
+			if i != 0xffffffff {
+				rule.SuppressPrefixlen = int(i)
+			}
+		case nl.FRA_SUPPRESS_IFGROUP:
+			i := native.Uint32(attrs[j].Value[0:4])
+			if i != 0xffffffff {
+				rule.SuppressIfgroup = int(i)
+			}
+		case nl.FRA_FLOW:
+			rule.Flow = int(native.Uint32(attrs[j].Value[0:4]))
+		case nl.FRA_GOTO:
+			rule.Goto = int(native.Uint32(attrs[j].Value[0:4]))
+		case nl.FRA_PRIORITY:
+			rule.Priority = int(native.Uint32(attrs[j].Value[0:4]))
+		case nl.FRA_IP_PROTO:
+			rule.IPProto = int(native.Uint32(attrs[j].Value[0:4]))
+		case nl.FRA_DPORT_RANGE:
+			rule.Dport = NewRulePortRange(native.Uint16(attrs[j].Value[0:2]), native.Uint16(attrs[j].Value[2:4]))
+		case nl.FRA_SPORT_RANGE:
+			rule.Sport = NewRulePortRange(native.Uint16(attrs[j].Value[0:2]), native.Uint16(attrs[j].Value[2:4]))
+		case nl.FRA_UID_RANGE:
+			rule.UIDRange = NewRuleUIDRange(native.Uint32(attrs[j].Value[0:4]), native.Uint32(attrs[j].Value[4:8]))
+		case nl.FRA_PROTOCOL:
+			rule.Protocol = uint8(attrs[j].Value[0])
+		}
+	}
+
+	return rule, nil
+}
+
+// RuleSubscribe takes a chan down which notifications will be sent
+// when rules are added or deleted. Close the 'done' chan to stop subscription.
+func RuleSubscribe(ch chan<- RuleUpdate, done <-chan struct{}) error {
+	return ruleSubscribeAt(netns.None(), netns.None(), ch, done, nil, false, 0, nil, false)
+}
+
+// RuleSubscribeAt works like RuleSubscribe plus it allows the caller
+// to choose the network namespace in which to subscribe (ns).
+func RuleSubscribeAt(ns netns.NsHandle, ch chan<- RuleUpdate, done <-chan struct{}) error {
+	return ruleSubscribeAt(ns, netns.None(), ch, done, nil, false, 0, nil, false)
+}
+
+// RuleSubscribeOptions contains a set of options to use with
+// RuleSubscribeWithOptions.
+type RuleSubscribeOptions struct {
+	Namespace              *netns.NsHandle
+	ErrorCallback          func(error)
+	ListExisting           bool
+	ReceiveBufferSize      int
+	ReceiveBufferForceSize bool
+	ReceiveTimeout         *unix.Timeval
+}
+
+// RuleSubscribeWithOptions work like RuleSubscribe but enable to
+// provide additional options to modify the behavior. Currently, the
+// namespace can be provided as well as an error callback.
+func RuleSubscribeWithOptions(ch chan<- RuleUpdate, done <-chan struct{}, options RuleSubscribeOptions) error {
+	if options.Namespace == nil {
+		none := netns.None()
+		options.Namespace = &none
+	}
+	return ruleSubscribeAt(*options.Namespace, netns.None(), ch, done, options.ErrorCallback, options.ListExisting,
+		options.ReceiveBufferSize, options.ReceiveTimeout, options.ReceiveBufferForceSize)
+}
+
+func ruleSubscribeAt(newNs, curNs netns.NsHandle, ch chan<- RuleUpdate, done <-chan struct{}, cberr func(error), listExisting bool,
+	rcvbuf int, rcvTimeout *unix.Timeval, rcvbufForce bool) error {
+	s, err := nl.SubscribeAt(newNs, curNs, unix.NETLINK_ROUTE, unix.RTNLGRP_IPV4_RULE, unix.RTNLGRP_IPV6_RULE)
+	if err != nil {
+		return err
+	}
+	if rcvTimeout != nil {
+		if err := s.SetReceiveTimeout(rcvTimeout); err != nil {
+			return err
+		}
+	}
+	if rcvbuf != 0 {
+		err = s.SetReceiveBufferSize(rcvbuf, rcvbufForce)
+		if err != nil {
+			return err
+		}
+	}
+	if done != nil {
+		go func() {
+			<-done
+			s.Close()
+		}()
+	}
+	if listExisting {
+		req := pkgHandle.newNetlinkRequest(unix.RTM_GETRULE,
+			unix.NLM_F_DUMP)
+		infmsg := nl.NewIfInfomsg(unix.AF_UNSPEC)
+		req.AddData(infmsg)
+		if err := s.Send(req); err != nil {
+			return err
+		}
+	}
+	go func() {
+		defer close(ch)
+		for {
+			msgs, from, err := s.Receive()
+			if err != nil {
+				if cberr != nil {
+					cberr(fmt.Errorf("Receive failed: %v",
+						err))
+				}
+				if err.Error() != "use of closed file" && err.Error() != "file already closed" {
+					panic(fmt.Sprintf("err: %v", err))
+				}
+				return
+			}
+			if from.Pid != nl.PidKernel {
+				if cberr != nil {
+					cberr(fmt.Errorf("Wrong sender portid %d, expected %d", from.Pid, nl.PidKernel))
+				}
+				panic(fmt.Sprintf("from.Pid: %d, nl.PidKernel: %d", from.Pid, nl.PidKernel))
+				continue
+			}
+			for _, m := range msgs {
+				if m.Header.Type == unix.NLMSG_DONE {
+					continue
+				}
+				if m.Header.Type == unix.NLMSG_ERROR {
+					error := int32(native.Uint32(m.Data[0:4]))
+					if error == 0 {
+						continue
+					}
+					if cberr != nil {
+						cberr(fmt.Errorf("error message: %v",
+							syscall.Errno(-error)))
+					}
+					continue
+				}
+				rule, err := deserializeRule(m.Data)
+				if err != nil {
+					if cberr != nil {
+						cberr(err)
+					}
+					continue
+				}
+				ch <- RuleUpdate{Type: m.Header.Type, Rule: *rule}
+			}
+		}
+	}()
+
+	return nil
+}