Feature/systemd query events (#1728)

* [service] Subscribe to systemd-resolver events

* [service] Add disabled state to the resolver

* [service] Add ETW DNS event listener

* [service] DNS listener refactoring

* [service] Add windows core dll project

* [service] DNSListener refactoring, small bugfixes

* [service] Change dns bypass rule

* [service] Update gitignore

* [service] Remove shim from integration module

* [service] Add DNS packet analyzer

* [service] Add self-check in dns monitor

* [service] Fix go linter errors

* [CI] Add github workflow for the windows core dll

* [service] Minor fixes to the dns monitor

service/firewall/interception/dnsmonitor/etwlink_windows.go

@@ -0,0 +1,99 @@
+//go:build windows
+// +build windows
+
+package dnsmonitor
+
+import (
+	"fmt"
+	"runtime"
+	"sync"
+	"sync/atomic"
+
+	"github.com/safing/portmaster/service/integration"
+	"golang.org/x/sys/windows"
+)
+
+type ETWSession struct {
+	i integration.ETWFunctions
+
+	shutdownGuard atomic.Bool
+	shutdownMutex sync.Mutex
+
+	state uintptr
+}
+
+// NewSession creates new ETW event listener and initilizes it. This is a low level interface, make sure to call DestorySession when you are done using it.
+func NewSession(etwInterface integration.ETWFunctions, callback func(domain string, result string)) (*ETWSession, error) {
+	etwSession := &ETWSession{
+		i: etwInterface,
+	}
+
+	// Make sure session from previous instances are not running.
+	_ = etwSession.i.StopOldSession()
+
+	// Initialize notification activated callback
+	win32Callback := windows.NewCallback(func(domain *uint16, result *uint16) uintptr {
+		callback(windows.UTF16PtrToString(domain), windows.UTF16PtrToString(result))
+		return 0
+	})
+	// The function only allocates memory it will not fail.
+	etwSession.state = etwSession.i.CreateState(win32Callback)
+
+	// Make sure DestroySession is called even if caller forgets to call it.
+	runtime.SetFinalizer(etwSession, func(s *ETWSession) {
+		_ = s.i.DestroySession(s.state)
+	})
+
+	// Initialize session.
+	err := etwSession.i.InitializeSession(etwSession.state)
+	if err != nil {
+		return nil, fmt.Errorf("failed to initialzie session: %q", err)
+	}
+
+	return etwSession, nil
+}
+
+// StartTrace starts the tracing session of dns events. This is a blocking call. It will not return until the trace is stopped.
+func (l *ETWSession) StartTrace() error {
+	return l.i.StartTrace(l.state)
+}
+
+// IsRunning returns true if DestroySession has NOT been called.
+func (l *ETWSession) IsRunning() bool {
+	return !l.shutdownGuard.Load()
+}
+
+// FlushTrace flushes the trace buffer.
+func (l *ETWSession) FlushTrace() error {
+	l.shutdownMutex.Lock()
+	defer l.shutdownMutex.Unlock()
+
+	// Make sure session is still running.
+	if l.shutdownGuard.Load() {
+		return nil
+	}
+
+	return l.i.FlushTrace(l.state)
+}
+
+// StopTrace stopes the trace. This will cause StartTrace to return.
+func (l *ETWSession) StopTrace() error {
+	return l.i.StopTrace(l.state)
+}
+
+// DestroySession closes the session and frees the allocated memory. Listener cannot be used after this function is called.
+func (l *ETWSession) DestroySession() error {
+	l.shutdownMutex.Lock()
+	defer l.shutdownMutex.Unlock()
+
+	if l.shutdownGuard.Swap(true) {
+		return nil
+	}
+
+	err := l.i.DestroySession(l.state)
+	if err != nil {
+		return err
+	}
+	l.state = 0
+	return nil
+}

service/firewall/interception/dnsmonitor/eventlistener.go

@@ -0,0 +1,19 @@
+//go:build !linux && !windows
+// +build !linux,!windows
+
+package dnsmonitor
+
+type Listener struct{}
+
+func newListener(_ *DNSMonitor) (*Listener, error) {
+	return &Listener{}, nil
+}
+
+func (l *Listener) flush() error {
+	// Nothing to flush
+	return nil
+}
+
+func (l *Listener) stop() error {
+	return nil
+}

service/firewall/interception/dnsmonitor/eventlistener_linux.go

@@ -0,0 +1,144 @@
+//go:build linux
+// +build linux
+
+package dnsmonitor
+
+import (
+	"errors"
+	"fmt"
+	"net"
+	"os"
+
+	"github.com/miekg/dns"
+	"github.com/safing/portmaster/base/log"
+	"github.com/safing/portmaster/service/mgr"
+	"github.com/safing/portmaster/service/resolver"
+	"github.com/varlink/go/varlink"
+)
+
+type Listener struct {
+	varlinkConn *varlink.Connection
+}
+
+func newListener(module *DNSMonitor) (*Listener, error) {
+	// Set source of the resolver.
+	ResolverInfo.Source = resolver.ServerSourceSystemd
+
+	// Check if the system has systemd-resolver.
+	_, err := os.Stat("/run/systemd/resolve/io.systemd.Resolve.Monitor")
+	if err != nil {
+		return nil, fmt.Errorf("system does not support systemd resolver monitor")
+	}
+
+	listener := &Listener{}
+
+	restartAttempts := 0
+
+	module.mgr.Go("systemd-resolver-event-listener", func(w *mgr.WorkerCtx) error {
+		// Abort initialization if the connection failed after too many tries.
+		if restartAttempts > 10 {
+			return nil
+		}
+		restartAttempts += 1
+
+		// Initialize varlink connection
+		varlinkConn, err := varlink.NewConnection(module.mgr.Ctx(), "unix:/run/systemd/resolve/io.systemd.Resolve.Monitor")
+		if err != nil {
+			return fmt.Errorf("failed to connect to systemd-resolver varlink service: %w", err)
+		}
+		defer func() {
+			if varlinkConn != nil {
+				err = varlinkConn.Close()
+				if err != nil {
+					log.Errorf("dnsmonitor: failed to close varlink connection: %s", err)
+				}
+			}
+		}()
+
+		listener.varlinkConn = varlinkConn
+		// Subscribe to the dns query events
+		receive, err := listener.varlinkConn.Send(w.Ctx(), "io.systemd.Resolve.Monitor.SubscribeQueryResults", nil, varlink.More)
+		if err != nil {
+			var varlinkErr *varlink.Error
+			if errors.As(err, &varlinkErr) {
+				return fmt.Errorf("failed to issue Varlink call: %+v", varlinkErr.Parameters)
+			} else {
+				return fmt.Errorf("failed to issue Varlink call: %w", err)
+			}
+		}
+
+		for {
+			queryResult := QueryResult{}
+			// Receive the next event from the resolver.
+			flags, err := receive(w.Ctx(), &queryResult)
+			if err != nil {
+				var varlinkErr *varlink.Error
+				if errors.As(err, &varlinkErr) {
+					return fmt.Errorf("failed to receive Varlink reply: %+v", varlinkErr.Parameters)
+				} else {
+					return fmt.Errorf("failed to receive Varlink reply: %w", err)
+				}
+			}
+
+			// Check if the reply indicates the end of the stream
+			if flags&varlink.Continues == 0 {
+				break
+			}
+
+			// Ignore if there is no question.
+			if queryResult.Question == nil || len(*queryResult.Question) == 0 {
+				continue
+			}
+
+			// Protmaster self check
+			domain := (*queryResult.Question)[0].Name
+			if processIfSelfCheckDomain(dns.Fqdn(domain)) {
+				// Not need to process result.
+				continue
+			}
+
+			if queryResult.Rcode != nil {
+				continue // Ignore DNS errors
+			}
+
+			listener.processAnswer(domain, &queryResult)
+		}
+		return nil
+	})
+	return listener, nil
+}
+
+func (l *Listener) flush() error {
+	// Nothing to flush
+	return nil
+}
+
+func (l *Listener) stop() error {
+	return nil
+}
+
+func (l *Listener) processAnswer(domain string, queryResult *QueryResult) {
+	// Allocated data struct for the parsed result.
+	cnames := make(map[string]string)
+	ips := make([]net.IP, 0, 5)
+
+	// Check if the query is valid
+	if queryResult.Answer == nil {
+		return
+	}
+
+	// Go trough each answer entry.
+	for _, a := range *queryResult.Answer {
+		if a.RR.Address != nil {
+			ip := net.IP(*a.RR.Address)
+			// Answer contains ip address.
+			ips = append(ips, ip)
+
+		} else if a.RR.Name != nil {
+			// Answer is a CNAME.
+			cnames[domain] = *a.RR.Name
+		}
+	}
+
+	saveDomain(domain, ips, cnames)
+}

service/firewall/interception/dnsmonitor/eventlistener_windows.go

@@ -0,0 +1,103 @@
+//go:build windows
+// +build windows
+
+package dnsmonitor
+
+import (
+	"fmt"
+	"net"
+	"strconv"
+	"strings"
+
+	"github.com/miekg/dns"
+	"github.com/safing/portmaster/service/mgr"
+	"github.com/safing/portmaster/service/resolver"
+)
+
+type Listener struct {
+	etw *ETWSession
+}
+
+func newListener(module *DNSMonitor) (*Listener, error) {
+	// Set source of the resolver.
+	ResolverInfo.Source = resolver.ServerSourceETW
+
+	listener := &Listener{}
+	var err error
+	// Initialize new dns event session.
+	listener.etw, err = NewSession(module.instance.OSIntegration().GetETWInterface(), listener.processEvent)
+	if err != nil {
+		return nil, err
+	}
+
+	// Start listening for events.
+	module.mgr.Go("etw-dns-event-listener", func(w *mgr.WorkerCtx) error {
+		return listener.etw.StartTrace()
+	})
+
+	return listener, nil
+}
+
+func (l *Listener) flush() error {
+	return l.etw.FlushTrace()
+}
+
+func (l *Listener) stop() error {
+	if l == nil {
+		return fmt.Errorf("listener is nil")
+	}
+	if l.etw == nil {
+		return fmt.Errorf("invalid etw session")
+	}
+	// Stop and destroy trace. Destroy should be called even if stop fails for some reason.
+	err := l.etw.StopTrace()
+	err2 := l.etw.DestroySession()
+
+	if err != nil {
+		return fmt.Errorf("StopTrace failed: %w", err)
+	}
+
+	if err2 != nil {
+		return fmt.Errorf("DestroySession failed: %w", err2)
+	}
+	return nil
+}
+
+func (l *Listener) processEvent(domain string, result string) {
+	if processIfSelfCheckDomain(dns.Fqdn(domain)) {
+		// Not need to process result.
+		return
+	}
+
+	// Ignore empty results
+	if len(result) == 0 {
+		return
+	}
+
+	cnames := make(map[string]string)
+	ips := []net.IP{}
+
+	resultArray := strings.Split(result, ";")
+	for _, r := range resultArray {
+		// For results other than IP addresses, the string starts with "type:"
+		if strings.HasPrefix(r, "type:") {
+			dnsValueArray := strings.Split(r, " ")
+			if len(dnsValueArray) < 3 {
+				continue
+			}
+
+			// Ignore everything except CNAME records
+			if value, err := strconv.ParseInt(dnsValueArray[1], 10, 16); err == nil && value == int64(dns.TypeCNAME) {
+				cnames[domain] = dnsValueArray[2]
+			}
+
+		} else {
+			// If the event doesn't start with "type:", it's an IP address
+			ip := net.ParseIP(r)
+			if ip != nil {
+				ips = append(ips, ip)
+			}
+		}
+	}
+	saveDomain(domain, ips, cnames)
+}

service/firewall/interception/dnsmonitor/module.go

@@ -0,0 +1,138 @@
+package dnsmonitor
+
+import (
+	"errors"
+	"net"
+	"strings"
+
+	"github.com/miekg/dns"
+	"github.com/safing/portmaster/base/database"
+	"github.com/safing/portmaster/base/log"
+	"github.com/safing/portmaster/service/compat"
+	"github.com/safing/portmaster/service/integration"
+	"github.com/safing/portmaster/service/mgr"
+	"github.com/safing/portmaster/service/network/netutils"
+	"github.com/safing/portmaster/service/resolver"
+)
+
+var ResolverInfo = resolver.ResolverInfo{
+	Name: "SystemResolver",
+	Type: resolver.ServerTypeMonitor,
+}
+
+type DNSMonitor struct {
+	instance instance
+	mgr      *mgr.Manager
+
+	listener *Listener
+}
+
+// Manager returns the module manager.
+func (dl *DNSMonitor) Manager() *mgr.Manager {
+	return dl.mgr
+}
+
+// Start starts the module.
+func (dl *DNSMonitor) Start() error {
+	// Initialize dns event listener
+	var err error
+	dl.listener, err = newListener(dl)
+	if err != nil {
+		log.Errorf("dnsmonitor: failed to start dns listener: %s", err)
+	}
+
+	return nil
+}
+
+// Stop stops the module.
+func (dl *DNSMonitor) Stop() error {
+	if dl.listener != nil {
+		err := dl.listener.stop()
+		if err != nil {
+			log.Errorf("dnsmonitor: failed to close listener: %s", err)
+		}
+	}
+	return nil
+}
+
+// Flush flushes the buffer forcing all events to be processed.
+func (dl *DNSMonitor) Flush() error {
+	return dl.listener.flush()
+}
+
+func saveDomain(domain string, ips []net.IP, cnames map[string]string) {
+	fqdn := dns.Fqdn(domain)
+	// Create new record for this IP.
+	record := resolver.ResolvedDomain{
+		Domain:            fqdn,
+		Resolver:          &ResolverInfo,
+		DNSRequestContext: &resolver.DNSRequestContext{},
+		Expires:           0,
+	}
+
+	// Process cnames
+	record.AddCNAMEs(cnames)
+
+	// Add to cache
+	saveIPsInCache(ips, resolver.IPInfoProfileScopeGlobal, record)
+}
+
+func New(instance instance) (*DNSMonitor, error) {
+	// Initialize module
+	m := mgr.New("DNSMonitor")
+	module := &DNSMonitor{
+		mgr:      m,
+		instance: instance,
+	}
+
+	return module, nil
+}
+
+type instance interface {
+	OSIntegration() *integration.OSIntegration
+}
+
+func processIfSelfCheckDomain(fqdn string) bool {
+	// Check for compat check dns request.
+	if strings.HasSuffix(fqdn, compat.DNSCheckInternalDomainScope) {
+		subdomain := strings.TrimSuffix(fqdn, compat.DNSCheckInternalDomainScope)
+		_ = compat.SubmitDNSCheckDomain(subdomain)
+		log.Infof("dnsmonitor: self-check domain received")
+		// No need to parse the answer.
+		return true
+	}
+
+	return false
+}
+
+// saveIPsInCache saves the provided ips in the dns cashe assoseted with the record Domain and CNAMEs.
+func saveIPsInCache(ips []net.IP, profileID string, record resolver.ResolvedDomain) {
+	// Package IPs and CNAMEs into IPInfo structs.
+	for _, ip := range ips {
+		// Never save domain attributions for localhost IPs.
+		if netutils.GetIPScope(ip) == netutils.HostLocal {
+			continue
+		}
+
+		ipString := ip.String()
+		info, err := resolver.GetIPInfo(profileID, ipString)
+		if err != nil {
+			if !errors.Is(err, database.ErrNotFound) {
+				log.Errorf("dnsmonitor: failed to search for IP info record: %s", err)
+			}
+
+			info = &resolver.IPInfo{
+				IP:        ipString,
+				ProfileID: profileID,
+			}
+		}
+
+		// Add the new record to the resolved domains for this IP and scope.
+		info.AddDomain(record)
+
+		// Save if the record is new or has been updated.
+		if err := info.Save(); err != nil {
+			log.Errorf("dnsmonitor: failed to save IP info record: %s", err)
+		}
+	}
+}

service/firewall/interception/dnsmonitor/varlinktypes.go

@@ -0,0 +1,83 @@
+//go:build linux
+// +build linux
+
+package dnsmonitor
+
+// List of struct that define the systemd-resolver varlink dns event protocol.
+// Source: `sudo varlinkctl introspect /run/systemd/resolve/io.systemd.Resolve.Monitor io.systemd.Resolve.Monitor`
+
+type ResourceKey struct {
+	Class int    `json:"class"`
+	Type  int    `json:"type"`
+	Name  string `json:"name"`
+}
+
+type ResourceRecord struct {
+	Key     ResourceKey `json:"key"`
+	Name    *string     `json:"name,omitempty"`
+	Address *[]byte     `json:"address,omitempty"`
+	// Rest of the fields are not used.
+	// Priority     *int        `json:"priority,omitempty"`
+	// Weight       *int        `json:"weight,omitempty"`
+	// Port         *int        `json:"port,omitempty"`
+	// CPU          *string     `json:"cpu,omitempty"`
+	// OS           *string     `json:"os,omitempty"`
+	// Items        *[]string   `json:"items,omitempty"`
+	// MName        *string     `json:"mname,omitempty"`
+	// RName        *string     `json:"rname,omitempty"`
+	// Serial       *int        `json:"serial,omitempty"`
+	// Refresh      *int        `json:"refresh,omitempty"`
+	// Expire       *int        `json:"expire,omitempty"`
+	// Minimum      *int        `json:"minimum,omitempty"`
+	// Exchange     *string     `json:"exchange,omitempty"`
+	// Version      *int        `json:"version,omitempty"`
+	// Size         *int        `json:"size,omitempty"`
+	// HorizPre     *int        `json:"horiz_pre,omitempty"`
+	// VertPre      *int        `json:"vert_pre,omitempty"`
+	// Latitude     *int        `json:"latitude,omitempty"`
+	// Longitude    *int        `json:"longitude,omitempty"`
+	// Altitude     *int        `json:"altitude,omitempty"`
+	// KeyTag       *int        `json:"key_tag,omitempty"`
+	// Algorithm    *int        `json:"algorithm,omitempty"`
+	// DigestType   *int        `json:"digest_type,omitempty"`
+	// Digest       *string     `json:"digest,omitempty"`
+	// FPType       *int        `json:"fptype,omitempty"`
+	// Fingerprint  *string     `json:"fingerprint,omitempty"`
+	// Flags        *int        `json:"flags,omitempty"`
+	// Protocol     *int        `json:"protocol,omitempty"`
+	// DNSKey       *string     `json:"dnskey,omitempty"`
+	// Signer       *string     `json:"signer,omitempty"`
+	// TypeCovered  *int        `json:"type_covered,omitempty"`
+	// Labels       *int        `json:"labels,omitempty"`
+	// OriginalTTL  *int        `json:"original_ttl,omitempty"`
+	// Expiration   *int        `json:"expiration,omitempty"`
+	// Inception    *int        `json:"inception,omitempty"`
+	// Signature    *string     `json:"signature,omitempty"`
+	// NextDomain   *string     `json:"next_domain,omitempty"`
+	// Types        *[]int      `json:"types,omitempty"`
+	// Iterations   *int        `json:"iterations,omitempty"`
+	// Salt         *string     `json:"salt,omitempty"`
+	// Hash         *string     `json:"hash,omitempty"`
+	// CertUsage    *int        `json:"cert_usage,omitempty"`
+	// Selector     *int        `json:"selector,omitempty"`
+	// MatchingType *int        `json:"matching_type,omitempty"`
+	// Data         *string     `json:"data,omitempty"`
+	// Tag          *string     `json:"tag,omitempty"`
+	// Value        *string     `json:"value,omitempty"`
+}
+
+type Answer struct {
+	RR      *ResourceRecord `json:"rr,omitempty"`
+	Raw     string          `json:"raw"`
+	IfIndex *int            `json:"ifindex,omitempty"`
+}
+
+type QueryResult struct {
+	Ready              *bool          `json:"ready,omitempty"`
+	State              *string        `json:"state,omitempty"`
+	Rcode              *int           `json:"rcode,omitempty"`
+	Errno              *int           `json:"errno,omitempty"`
+	Question           *[]ResourceKey `json:"question,omitempty"`
+	CollectedQuestions *[]ResourceKey `json:"collectedQuestions,omitempty"`
+	Answer             *[]Answer      `json:"answer,omitempty"`
+}

service/firewall/interception/nfq/nfq.go

@@ -188,7 +188,7 @@ func (q *Queue) packetHandler(ctx context.Context) func(nfqueue.Attribute) int {
 			return 0
 		}
 
-		if err := pmpacket.Parse(*attrs.Payload, &pkt.Base); err != nil {
+		if err := pmpacket.ParseLayer3(*attrs.Payload, &pkt.Base); err != nil {
 			log.Warningf("nfqueue: failed to parse payload: %s", err)
 			_ = pkt.Drop()
 			return 0

service/firewall/interception/windowskext/packet.go

@@ -59,7 +59,7 @@ func (pkt *Packet) LoadPacketData() error {
 			return packet.ErrFailedToLoadPayload
 		}
 
-		err = packet.Parse(payload, &pkt.Base)
+		err = packet.ParseLayer3(payload, &pkt.Base)
 		if err != nil {
 			log.Tracer(pkt.Ctx()).Warningf("windowskext: failed to parse payload: %s", err)
 			return packet.ErrFailedToLoadPayload

service/firewall/interception/windowskext2/handler.go

@@ -55,6 +55,7 @@ func Handler(ctx context.Context, packets chan packet.Packet, bandwidthUpdate ch
 				newPacket := &Packet{
 					verdictRequest: conn.ID,
 					payload:        conn.Payload,
+					payloadLayer:   conn.PayloadLayer,
 					verdictSet:     abool.NewBool(false),
 				}
 				info := newPacket.Info()

service/firewall/interception/windowskext2/packet.go

@@ -4,6 +4,7 @@
 package windowskext
 
 import (
+	"fmt"
 	"sync"
 
 	"github.com/tevino/abool"
@@ -19,6 +20,7 @@ type Packet struct {
 
 	verdictRequest uint64
 	payload        []byte
+	payloadLayer   uint8
 	verdictSet     *abool.AtomicBool
 
 	payloadLoaded bool
@@ -51,7 +53,15 @@ func (pkt *Packet) LoadPacketData() error {
 		pkt.payloadLoaded = true
 
 		if len(pkt.payload) > 0 {
-			err := packet.Parse(pkt.payload, &pkt.Base)
+			var err error
+			switch pkt.payloadLayer {
+			case 3:
+				err = packet.ParseLayer3(pkt.payload, &pkt.Base)
+			case 4:
+				err = packet.ParseLayer4(pkt.payload, &pkt.Base)
+			default:
+				err = fmt.Errorf("unsupported payload layer: %d", pkt.payloadLayer)
+			}
 			if err != nil {
 				log.Tracef("payload: %#v", pkt.payload)
 				log.Tracer(pkt.Ctx()).Warningf("windowskext: failed to parse payload: %s", err)