Add experimental nfqueue interception backend

This commit adds a new experimental nfqueue interception
backed based on github.com/florianl/go-nfqueue. It should
add more stability to the interception of network packets
and also eliminates the CGO and the runtime dependency on
libnetfilter_queue. Note that this commit does not remove
the old nfqueue backend yet but adds a --experimental-nfqueue
flag to portmaster-core.

cmds/portmaster-core/.gitignore

@@ -1,5 +1,6 @@
 # Compiled binaries
 portmaster
+portmaster-core
 portmaster.exe
 dnsonly
 dnsonly.exe

firewall/interception/nfqexp/nfqexp.go

@@ -0,0 +1,122 @@
+// Package nfqexp contains a nfqueue library experiment.
+package nfqexp
+
+import (
+	"context"
+	"time"
+
+	"github.com/safing/portbase/log"
+	pmpacket "github.com/safing/portmaster/network/packet"
+	"golang.org/x/sys/unix"
+
+	"github.com/florianl/go-nfqueue"
+)
+
+// Queue wraps a nfqueue
+type Queue struct {
+	id                   uint16
+	nf                   *nfqueue.Nfqueue
+	packets              chan pmpacket.Packet
+	cancelSocketCallback context.CancelFunc
+}
+
+// New opens a new nfQueue.
+func New(qid uint16, v6 bool) (*Queue, error) {
+	afFamily := unix.AF_INET
+	if v6 {
+		afFamily = unix.AF_INET6
+	}
+	cfg := &nfqueue.Config{
+		NfQueue:      qid,
+		MaxPacketLen: 0xffff,
+		MaxQueueLen:  0xff,
+		AfFamily:     uint8(afFamily),
+		Copymode:     nfqueue.NfQnlCopyPacket,
+		ReadTimeout:  50 * time.Millisecond,
+		WriteTimeout: 50 * time.Millisecond,
+	}
+
+	nf, err := nfqueue.Open(cfg)
+	if err != nil {
+		return nil, err
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	q := &Queue{
+		id:                   qid,
+		nf:                   nf,
+		packets:              make(chan pmpacket.Packet, 1000),
+		cancelSocketCallback: cancel,
+	}
+
+	fn := func(attrs nfqueue.Attribute) int {
+
+		if attrs.PacketID == nil {
+			// we need a packet id to set a verdict,
+			// if we don't get an ID there's hardly anything
+			// we can do.
+			return 0
+		}
+
+		pkt := &packet{
+			ID:         *attrs.PacketID,
+			queue:      q,
+			received:   time.Now(),
+			verdictSet: make(chan struct{}),
+		}
+
+		if attrs.Payload != nil {
+			pkt.Payload = *attrs.Payload
+		}
+
+		if err := pmpacket.Parse(pkt.Payload, &pkt.Base); err != nil {
+			log.Warningf("nfqexp: failed to parse payload: %s", err)
+			_ = pkt.Drop()
+			return 0
+		}
+
+		select {
+		case q.packets <- pkt:
+			log.Tracef("nfqexp: queued packet %d (%s -> %s) after %s", pkt.ID, pkt.Info().Src, pkt.Info().Dst, time.Since(pkt.received))
+		case <-ctx.Done():
+			return 0
+		case <-time.After(time.Second):
+			log.Warningf("nfqexp: failed to queue packet (%s since it was handed over by the kernel)", time.Since(pkt.received))
+		}
+
+		go func() {
+			select {
+			case <-pkt.verdictSet:
+
+			case <-time.After(5 * time.Second):
+				log.Warningf("nfqexp: no verdict set for packet %d (%s -> %s) after %s, dropping", pkt.ID, pkt.Info().Src, pkt.Info().Dst, time.Since(pkt.received))
+				if err := pkt.Drop(); err != nil {
+					log.Warningf("nfqexp: failed to apply default-drop to unveridcted packet %d (%s -> %s)", pkt.ID, pkt.Info().Src, pkt.Info().Dst)
+				}
+			}
+		}()
+
+		return 0 // continue calling this fn
+	}
+
+	if err := q.nf.Register(ctx, fn); err != nil {
+		defer q.nf.Close()
+		return nil, err
+	}
+
+	return q, nil
+}
+
+// Destroy destroys the queue. Any error encountered is logged.
+func (q *Queue) Destroy() {
+	q.cancelSocketCallback()
+
+	if err := q.nf.Close(); err != nil {
+		log.Errorf("nfqexp: failed to close queue %d: %s", q.id, err)
+	}
+}
+
+// PacketChannel returns the packet channel.
+func (q *Queue) PacketChannel() <-chan pmpacket.Packet {
+	return q.packets
+}

firewall/interception/nfqexp/packet.go

@@ -0,0 +1,122 @@
+package nfqexp
+
+import (
+	"errors"
+	"time"
+
+	"github.com/florianl/go-nfqueue"
+	"github.com/mdlayher/netlink"
+	"github.com/safing/portbase/log"
+	pmpacket "github.com/safing/portmaster/network/packet"
+)
+
+// Firewalling marks used by the Portmaster.
+// See TODO on packet.mark() on their relevance
+// and a possibility to remove most IPtables rules.
+const (
+	MarkAccept       = 1700
+	MarkBlock        = 1701
+	MarkDrop         = 1702
+	MarkAcceptAlways = 1710
+	MarkBlockAlways  = 1711
+	MarkDropAlways   = 1712
+	MarkRerouteNS    = 1799
+	MarkRerouteSPN   = 1717
+)
+
+func markToString(mark int) string {
+	switch mark {
+	case MarkAccept:
+		return "Accept"
+	case MarkBlock:
+		return "Block"
+	case MarkDrop:
+		return "Drop"
+	case MarkAcceptAlways:
+		return "AcceptAlways"
+	case MarkBlockAlways:
+		return "BlockAlways"
+	case MarkDropAlways:
+		return "DropAlways"
+	case MarkRerouteNS:
+		return "RerouteNS"
+	case MarkRerouteSPN:
+		return "RerouteSPN"
+	}
+	return "unknown"
+}
+
+// packet implements the packet.Packet interface.
+type packet struct {
+	pmpacket.Base
+	ID         uint32
+	received   time.Time
+	queue      *Queue
+	verdictSet chan struct{}
+}
+
+// TODO(ppacher): revisit the following behavior:
+// 		The legacy implementation of nfqueue (and the interception) module
+// 		always accept a packet but may mark it so that a subsequent rule in
+// 		the C17 chain drops, rejects or modifies it.
+//
+//		For drop/return we could use the actual nfQueue verdicts Drop and Stop.
+//		Re-routing to local NS or SPN can be done by modifying the packet here
+//		and using SetVerdictModPacket and reject can be implemented using a simple
+// 		raw-socket.
+//
+func (pkt *packet) mark(mark int) (err error) {
+	defer func() {
+		if x := recover(); x != nil {
+			err = errors.New("verdict set")
+		}
+	}()
+	for {
+		if err := pkt.queue.nf.SetVerdictWithMark(pkt.ID, nfqueue.NfAccept, mark); err != nil {
+			log.Warningf("nfqexp: failed to set verdict %s for %d (%s -> %s): %s", markToString(mark), pkt.ID, pkt.Info().Src, pkt.Info().Dst, err)
+			if opErr, ok := err.(*netlink.OpError); ok {
+				if opErr.Timeout() || opErr.Temporary() {
+					continue
+				}
+			}
+
+			return err
+		}
+		break
+	}
+	log.Tracef("nfqexp: marking packet %d (%s -> %s) on queue %d with %s after %s", pkt.ID, pkt.Info().Src, pkt.Info().Dst, pkt.queue.id, markToString(mark), time.Since(pkt.received))
+	close(pkt.verdictSet)
+	return nil
+}
+
+func (pkt *packet) Accept() error {
+	return pkt.mark(MarkAccept)
+}
+
+func (pkt *packet) Block() error {
+	return pkt.mark(MarkBlock)
+}
+
+func (pkt *packet) Drop() error {
+	return pkt.mark(MarkDrop)
+}
+
+func (pkt *packet) PermanentAccept() error {
+	return pkt.mark(MarkAcceptAlways)
+}
+
+func (pkt *packet) PermanentBlock() error {
+	return pkt.mark(MarkBlockAlways)
+}
+
+func (pkt *packet) PermanentDrop() error {
+	return pkt.mark(MarkDropAlways)
+}
+
+func (pkt *packet) RerouteToNameserver() error {
+	return pkt.mark(MarkRerouteNS)
+}
+
+func (pkt *packet) RerouteToTunnel() error {
+	return pkt.mark(MarkRerouteSPN)
+}

firewall/interception/nfqueue/nfqueue.go

@@ -66,6 +66,11 @@ func NewNFQueue(qid uint16) (nfq *NFQueue, err error) {
 	return nfq, nil
 }
 
+// PacketChannel returns a packet channel
+func (nfq *NFQueue) PacketChannel() <-chan packet.Packet {
+	return nfq.Packets
+}
+
 func (nfq *NFQueue) init() error {
 	var err error
 	if nfq.h, err = C.nfq_open(); err != nil || nfq.h == nil {

firewall/interception/nfqueue/packet.go

@@ -44,7 +44,7 @@ type Packet struct {
 //     pkt.QueueID, pkt.Id, pkt.Protocol, pkt.Src, pkt.SrcPort, pkt.Dst, pkt.DstPort, pkt.Mark, pkt.Checksum, pkt.Tos, pkt.TTL)
 // }
 
-//nolint:unparam // FIXME
+// nolint:unparam
 func (pkt *Packet) setVerdict(v uint32) (err error) {
 	defer func() {
 		if x := recover(); x != nil {

firewall/interception/nfqueue_linux.go

@@ -1,13 +1,17 @@
 package interception
 
 import (
+	"flag"
 	"fmt"
 	"sort"
 	"strings"
 
 	"github.com/coreos/go-iptables/iptables"
 
+	"github.com/safing/portbase/log"
+	"github.com/safing/portmaster/firewall/interception/nfqexp"
 	"github.com/safing/portmaster/firewall/interception/nfqueue"
+	"github.com/safing/portmaster/network/packet"
 )
 
 // iptables -A OUTPUT -p icmp -j", "NFQUEUE", "--queue-num", "1", "--queue-bypass
@@ -21,14 +25,29 @@ var (
 	v6rules  []string
 	v6once   []string
 
-	out4Queue *nfqueue.NFQueue
+	out4Queue nfQueue
-	in4Queue  *nfqueue.NFQueue
+	in4Queue  nfQueue
-	out6Queue *nfqueue.NFQueue
+	out6Queue nfQueue
-	in6Queue  *nfqueue.NFQueue
+	in6Queue  nfQueue
 
 	shutdownSignal = make(chan struct{})
+
+	experimentalNfqueueBackend bool
 )
 
+func init() {
+	flag.BoolVar(&experimentalNfqueueBackend, "experimental-nfqueue", false, "use experimental nfqueue packet")
+}
+
+// nfQueueFactoryFunc creates a new nfQueue with qid as the queue number.
+type nfQueueFactoryFunc func(qid uint16, v6 bool) (nfQueue, error)
+
+// nfQueue encapsulates nfQueue providers
+type nfQueue interface {
+	PacketChannel() <-chan packet.Packet
+	Destroy()
+}
+
 func init() {
 
 	v4chains = []string{
@@ -203,6 +222,16 @@ func deactivateIPTables(protocol iptables.Protocol, rules, chains []string) erro
 
 // StartNfqueueInterception starts the nfqueue interception.
 func StartNfqueueInterception() (err error) {
+	var nfQueueFactory nfQueueFactoryFunc = func(qid uint16, v6 bool) (nfQueue, error) {
+		return nfqueue.NewNFQueue(qid)
+	}
+
+	if experimentalNfqueueBackend {
+		log.Infof("nfqueue: using experimental nfqueue backend")
+		nfQueueFactory = func(qid uint16, v6 bool) (nfQueue, error) {
+			return nfqexp.New(qid, v6)
+		}
+	}
 
 	err = activateNfqueueFirewall()
 	if err != nil {
@@ -210,25 +239,25 @@ func StartNfqueueInterception() (err error) {
 		return fmt.Errorf("could not initialize nfqueue: %s", err)
 	}
 
-	out4Queue, err = nfqueue.NewNFQueue(17040)
+	out4Queue, err = nfQueueFactory(17040, false)
 	if err != nil {
 		_ = Stop()
-		return fmt.Errorf("interception: failed to create nfqueue(IPv4, in): %s", err)
+		return fmt.Errorf("nfqueue(IPv4, out): %w", err)
 	}
-	in4Queue, err = nfqueue.NewNFQueue(17140)
+	in4Queue, err = nfQueueFactory(17140, false)
 	if err != nil {
 		_ = Stop()
-		return fmt.Errorf("interception: failed to create nfqueue(IPv4, in): %s", err)
+		return fmt.Errorf("nfqueue(IPv4, in): %w", err)
 	}
-	out6Queue, err = nfqueue.NewNFQueue(17060)
+	out6Queue, err = nfQueueFactory(17060, true)
 	if err != nil {
 		_ = Stop()
-		return fmt.Errorf("interception: failed to create nfqueue(IPv4, in): %s", err)
+		return fmt.Errorf("nfqueue(IPv6, out): %w", err)
 	}
-	in6Queue, err = nfqueue.NewNFQueue(17160)
+	in6Queue, err = nfQueueFactory(17160, true)
 	if err != nil {
 		_ = Stop()
-		return fmt.Errorf("interception: failed to create nfqueue(IPv4, in): %s", err)
+		return fmt.Errorf("nfqueue(IPv6, in): %w", err)
 	}
 
 	go handleInterception()
@@ -265,16 +294,16 @@ func handleInterception() {
 		select {
 		case <-shutdownSignal:
 			return
-		case pkt := <-out4Queue.Packets:
+		case pkt := <-out4Queue.PacketChannel():
 			pkt.SetOutbound()
 			Packets <- pkt
-		case pkt := <-in4Queue.Packets:
+		case pkt := <-in4Queue.PacketChannel():
 			pkt.SetInbound()
 			Packets <- pkt
-		case pkt := <-out6Queue.Packets:
+		case pkt := <-out6Queue.PacketChannel():
 			pkt.SetOutbound()
 			Packets <- pkt
-		case pkt := <-in6Queue.Packets:
+		case pkt := <-in6Queue.PacketChannel():
 			pkt.SetInbound()
 			Packets <- pkt
 		}