Merge pull request #354 from themighty1/upstreammaster

do not block connection processing when GUI popup is active.

daemon/main.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"bytes"
 	"context"
 	"flag"
 	"fmt"
@@ -10,8 +11,8 @@ import (
 	"os/signal"
 	"runtime"
 	"runtime/pprof"
-	"sync"
 	"syscall"
+	"time"
 
 	"github.com/evilsocket/opensnitch/daemon/conman"
 	"github.com/evilsocket/opensnitch/daemon/core"
@@ -26,17 +27,17 @@ import (
 )
 
 var (
-	lock          sync.RWMutex
-	procmonMethod = ""
-	logFile       = ""
-	rulesPath     = "rules"
-	noLiveReload  = false
-	queueNum      = 0
-	workers       = 16
-	debug         = false
-	warning       = false
-	important     = false
-	errorlog      = false
+	procmonMethod  = ""
+	logFile        = ""
+	rulesPath      = "rules"
+	noLiveReload   = false
+	queueNum       = 0
+	repeatQueueNum int //will be set later to queueNum + 1
+	workers        = 16
+	debug          = false
+	warning        = false
+	important      = false
+	errorlog       = false
 
 	uiSocket = ""
 	uiClient = (*ui.Client)(nil)
@@ -44,16 +45,17 @@ var (
 	cpuProfile = ""
 	memProfile = ""
 
-	ctx      = (context.Context)(nil)
-	cancel   = (context.CancelFunc)(nil)
-	err      = (error)(nil)
-	rules    = (*rule.Loader)(nil)
-	stats    = (*statistics.Statistics)(nil)
-	queue    = (*netfilter.Queue)(nil)
-	pktChan  = (<-chan netfilter.Packet)(nil)
-	wrkChan  = (chan netfilter.Packet)(nil)
-	sigChan  = (chan os.Signal)(nil)
-	exitChan = (chan bool)(nil)
+	ctx           = (context.Context)(nil)
+	cancel        = (context.CancelFunc)(nil)
+	err           = (error)(nil)
+	rules         = (*rule.Loader)(nil)
+	stats         = (*statistics.Statistics)(nil)
+	queue         = (*netfilter.Queue)(nil)
+	repeatPktChan = (<-chan netfilter.Packet)(nil)
+	pktChan       = (<-chan netfilter.Packet)(nil)
+	wrkChan       = (chan netfilter.Packet)(nil)
+	sigChan       = (chan os.Signal)(nil)
+	exitChan      = (chan bool)(nil)
 )
 
 func init() {
@@ -148,12 +150,13 @@ func setupWorkers() {
 	}
 }
 
-func doCleanup(queue *netfilter.Queue) {
+func doCleanup(queue, repeatQueue *netfilter.Queue) {
 	log.Info("Cleaning up ...")
 	firewall.Stop(&queueNum)
 	procmon.End()
 	uiClient.Close()
 	queue.Close()
+	repeatQueue.Close()
 
 	if cpuProfile != "" {
 		pprof.StopCPUProfile()
@@ -212,53 +215,86 @@ func applyDefaultAction(packet *netfilter.Packet) {
 }
 
 func acceptOrDeny(packet *netfilter.Packet, con *conman.Connection) *rule.Rule {
-	lock.Lock()
-	defer lock.Unlock()
-
-	connected := false
 	r := rules.FindFirstMatch(con)
 	if r == nil {
-		// no rule matched, send a request to the
-		// UI client if connected and running
-		r, connected = uiClient.Ask(con)
+		// no rule matched
+		// Note that as soon as we set a verdict on a packet, the next packet in the netfilter queue
+		// will begin to be processed even if this function hasn't yet returned
+
+		// send a request to the UI client if
+		// 1) connected and running and 2) we are not already asking
+		if uiClient.Connected() == false || uiClient.GetIsAsking() == true {
+			applyDefaultAction(packet)
+			log.Debug("UI is not running or busy")
+			return nil
+		}
+
+		uiClient.SetIsAsking(true)
+		defer uiClient.SetIsAsking(false)
+
+		// In order not to block packet processing, we send our packet to a different netfilter queue
+		// and then immediately pull it back out of that queue
+		packet.SetRequeueVerdict(uint16(repeatQueueNum))
+
+		var o bool
+		var pkt netfilter.Packet
+		// don't wait for the packet longer than 1 sec
+		select {
+		case pkt, o = <-repeatPktChan:
+			if !o {
+				log.Debug("error while receiving packet from repeatPktChan")
+				return nil
+			}
+		case <-time.After(1 * time.Second):
+			log.Debug("timed out while receiving packet from repeatPktChan")
+			return nil
+		}
+
+		//check if the pulled out packet is the same we put in
+		if res := bytes.Compare(packet.Packet.Data(), pkt.Packet.Data()); res != 0 {
+			log.Error("The packet which was requeued has changed abruptly. This should never happen. Please report this incident to the Opensnitch developers. %s %s ", packet, pkt)
+			return nil
+		}
+		packet = &pkt
+
+		r = uiClient.Ask(con)
 		if r == nil {
 			log.Error("Invalid rule received, applying default action")
 			applyDefaultAction(packet)
 			return nil
 		}
-		if connected {
-			ok := false
-			pers := ""
-			action := string(r.Action)
-			if r.Action == rule.Allow {
-				action = log.Green(action)
-			} else {
-				action = log.Red(action)
-			}
+		ok := false
+		pers := ""
+		action := string(r.Action)
+		if r.Action == rule.Allow {
+			action = log.Green(action)
+		} else {
+			action = log.Red(action)
+		}
 
-			// check if and how the rule needs to be saved
-			if r.Duration == rule.Always {
-				pers = "Saved"
-				// add to the loaded rules and persist on disk
-				if err := rules.Add(r, true); err != nil {
-					log.Error("Error while saving rule: %s", err)
-				} else {
-					ok = true
-				}
+		// check if and how the rule needs to be saved
+		if r.Duration == rule.Always {
+			pers = "Saved"
+			// add to the loaded rules and persist on disk
+			if err := rules.Add(r, true); err != nil {
+				log.Error("Error while saving rule: %s", err)
 			} else {
-				pers = "Added"
-				// add to the rules but do not save to disk
-				if err := rules.Add(r, false); err != nil {
-					log.Error("Error while adding rule: %s", err)
-				} else {
-					ok = true
-				}
+				ok = true
 			}
-
-			if ok {
-				log.Important("%s new rule: %s if %s", pers, action, r.Operator.String())
+		} else {
+			pers = "Added"
+			// add to the rules but do not save to disk
+			if err := rules.Add(r, false); err != nil {
+				log.Error("Error while adding rule: %s", err)
+			} else {
+				ok = true
 			}
 		}
+
+		if ok {
+			log.Important("%s new rule: %s if %s", pers, action, r.Operator.String())
+		}
+
 	}
 
 	if r.Enabled == false {
@@ -270,7 +306,6 @@ func acceptOrDeny(packet *netfilter.Packet, con *conman.Connection) *rule.Rule {
 		if packet != nil {
 			packet.SetVerdictAndMark(netfilter.NF_ACCEPT, packet.Mark)
 		}
-
 		ruleName := log.Green(r.Name)
 		if r.Operator.Operand == rule.OpTrue {
 			ruleName = log.Dim(r.Name)
@@ -326,11 +361,19 @@ func main() {
 	setupWorkers()
 	queue, err := netfilter.NewQueue(uint16(queueNum))
 	if err != nil {
-		log.Warning("Is opnensitchd already running?")
+		log.Warning("Is opensnitchd already running?")
 		log.Fatal("Error while creating queue #%d: %s", queueNum, err)
 	}
 	pktChan = queue.Packets()
 
+	repeatQueueNum = queueNum + 1
+	repeatQueue, rqerr := netfilter.NewQueue(uint16(repeatQueueNum))
+	if rqerr != nil {
+		log.Warning("Is opensnitchd already running?")
+		log.Fatal("Error while creating queue #%d: %s", repeatQueueNum, rqerr)
+	}
+	repeatPktChan = repeatQueue.Packets()
+
 	uiClient = ui.NewClient(uiSocket, stats, rules)
 	if overwriteLogging() {
 		setupLogging()
@@ -359,6 +402,6 @@ func main() {
 	}
 Exit:
 	close(wrkChan)
-	doCleanup(queue)
+	doCleanup(queue, repeatQueue)
 	os.Exit(0)
 }

daemon/ui/client.go

@@ -56,6 +56,8 @@ type Client struct {
 	client              protocol.UIClient
 	configWatcher       *fsnotify.Watcher
 	streamNotifications protocol.UI_NotificationsClient
+	//isAsking is set to true if the client is awaiting a decision from the GUI
+	isAsking bool
 }
 
 // NewClient creates and configures a new client.
@@ -64,6 +66,7 @@ func NewClient(socketPath string, stats *statistics.Statistics, rules *rule.Load
 		stats:        stats,
 		rules:        rules,
 		isUnixSocket: false,
+		isAsking:     false,
 	}
 	c.clientCtx, c.clientCancel = context.WithCancel(context.Background())
 
@@ -116,14 +119,28 @@ func (c *Client) DefaultDuration() rule.Duration {
 
 // Connected checks if the client has established a connection with the server.
 func (c *Client) Connected() bool {
-	c.Lock()
-	defer c.Unlock()
+	c.RLock()
+	defer c.RUnlock()
 	if c.con == nil || c.con.GetState() != connectivity.Ready {
 		return false
 	}
 	return true
 }
 
+//GetIsAsking returns the isAsking flag
+func (c *Client) GetIsAsking() bool {
+	c.RLock()
+	defer c.RUnlock()
+	return c.isAsking
+}
+
+//SetIsAsking sets the isAsking flag
+func (c *Client) SetIsAsking(flag bool) {
+	c.Lock()
+	defer c.Unlock()
+	c.isAsking = flag
+}
+
 func (c *Client) poller() {
 	log.Debug("UI service poller started for socket %s", c.socketPath)
 	wasConnected := false
@@ -253,28 +270,21 @@ func (c *Client) ping(ts time.Time) (err error) {
 
 // Ask sends a request to the server, with the values of a connection to be
 // allowed or denied.
-func (c *Client) Ask(con *conman.Connection) (*rule.Rule, bool) {
-	if c.Connected() == false {
-		return clientDisconnectedRule, false
-	}
-
-	c.Lock()
-	defer c.Unlock()
-
+func (c *Client) Ask(con *conman.Connection) *rule.Rule {
 	// FIXME: if timeout is fired, the rule is not added to the list in the GUI
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second*120)
 	defer cancel()
 	reply, err := c.client.AskRule(ctx, con.Serialize())
 	if err != nil {
 		log.Warning("Error while asking for rule: %s - %v", err, con)
-		return nil, false
+		return nil
 	}
 
 	r, err := rule.Deserialize(reply)
 	if err != nil {
-		return nil, false
+		return nil
 	}
-	return r, true
+	return r
 }
 
 func (c *Client) monitorConfigWorker() {