feat(WIP): add pause and resume functionality for Portmaster/SPN

https://github.com/safing/portmaster/issues/2050

desktop/angular/projects/safing/portmaster-api/src/lib/portapi.service.ts

@@ -167,6 +167,23 @@ export class PortapiService {
     });
   }
 
+  /** Triggers a pause of the portmaster or SPN service
+   * @param duration The duration of the pause in seconds
+   * @param onlySPN Whether or not only the SPN should be paused
+   */
+  pause(duration: number, onlySPN: boolean): Observable<any> {
+    return this.http.post(`${this.httpEndpoint}/v1/control/pause`, { duration, onlySPN }, {
+      observe: 'response',
+      responseType: 'arraybuffer',
+    });
+  }
+  /** Triggers a resume of the portmaster (and SPN) service */
+  resume(): Observable<any> {
+    return this.http.post(`${this.httpEndpoint}/v1/control/resume`, undefined, {
+      observe: 'response',
+      responseType: 'arraybuffer',
+    });
+  }  
   /** Force the portmaster to check for updates */
   checkForUpdates(): Observable<any> {
     return this.http.post(`${this.httpEndpoint}/v1/updates/check`, undefined, {

service/compat/module.go

@@ -78,6 +78,8 @@ func prep() error {
 }
 
 func start() error {
+	isActive.Store(true)
+
 	startNotify()
 
 	selfcheckNetworkChangedFlag.Refresh()
@@ -92,13 +94,17 @@ func start() error {
 }
 
 func stop() error {
-	// selfcheckTask.Cancel()
-	// selfcheckTask = nil
+	isActive.Store(false)
+	resetSelfCheckState()
 
 	return nil
 }
 
 func selfcheckTaskFunc(wc *mgr.WorkerCtx) error {
+	if !isActive.Load() {
+		return nil
+	}
+
 	// Create tracing logger.
 	ctx, tracer := log.AddTracer(wc.Ctx())
 	defer tracer.Submit()
@@ -138,12 +144,16 @@ func selfcheckTaskFunc(wc *mgr.WorkerCtx) error {
 	}
 
 	// Reset self-check state.
+	resetSelfCheckState()
+
+	return nil
+}
+
+func resetSelfCheckState() {
 	selfcheckNetworkChangedFlag.Refresh()
 	selfCheckIsFailing.UnSet()
 	selfcheckFails = 0
 	resetSystemIssue()
-
-	return nil
 }
 
 // SelfCheckIsFailing returns whether the self check is currently failing.
@@ -156,6 +166,9 @@ func SelfCheckIsFailing() bool {
 var (
 	module     *Compat
 	shimLoaded atomic.Bool
+	// isActive is a simple shutdown flag to prevent self-check worker from executing during stop().
+	// TODO: consider correct start/stop of the workers instead.
+	isActive atomic.Bool
 )
 
 // New returns a new Compat module.

service/control/api.go

@@ -0,0 +1,49 @@
+package control
+
+import (
+	"net/http"
+
+	"github.com/safing/portmaster/base/api"
+)
+
+type pauseRequestParams struct {
+	Duration int  `json:"duration"` // Duration in seconds
+	OnlySPN  bool `json:"onlySPN"`  // Whether to pause only the SPN service
+}
+
+func (c *Control) registerAPIEndpoints() error {
+
+	if err := api.RegisterEndpoint(api.Endpoint{
+		Path:        "control/pause",
+		Write:       api.PermitAdmin,
+		ActionFunc:  c.handlePause,
+		Name:        "Pause Portmaster",
+		Description: "Pause the Portmaster Core Service.",
+		Parameters: []api.Parameter{
+			{
+				Method:      http.MethodPost,
+				Field:       "duration",
+				Description: "Specify the duration to pause the service in seconds.",
+			},
+			{
+				Method:      http.MethodPost,
+				Field:       "onlySPN",
+				Value:       "false",
+				Description: "Specify whether to pause only the SPN service.",
+			}},
+	}); err != nil {
+		return err
+	}
+
+	if err := api.RegisterEndpoint(api.Endpoint{
+		Path:        "control/resume",
+		Write:       api.PermitAdmin,
+		ActionFunc:  c.handleResume,
+		Name:        "Resume Portmaster",
+		Description: "Resume the Portmaster Core Service.",
+	}); err != nil {
+		return err
+	}
+
+	return nil
+}

service/control/module.go

@@ -0,0 +1,74 @@
+package control
+
+import (
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/safing/portmaster/base/config"
+	"github.com/safing/portmaster/service/compat"
+	"github.com/safing/portmaster/service/firewall/interception"
+	"github.com/safing/portmaster/service/mgr"
+)
+
+var logPrefix = "control: "
+
+type Control struct {
+	mgr      *mgr.Manager
+	instance instance
+
+	locker         sync.Mutex
+	pauseWorker    *mgr.WorkerMgr
+	isPaused       bool
+	isPausedSPN    bool
+	pauseStartTime time.Time
+	pauseDuration  time.Duration
+}
+
+type instance interface {
+	Config() *config.Config
+	Interception() *interception.Interception
+	Compat() *compat.Compat
+	SPNGroup() *mgr.ExtendedGroup
+}
+
+var (
+	singleton atomic.Bool
+)
+
+func New(instance instance) (*Control, error) {
+	if !singleton.CompareAndSwap(false, true) {
+		return nil, fmt.Errorf("control: New failed: instance already created")
+	}
+
+	mgr := mgr.New("Control")
+	module := &Control{
+		mgr:      mgr,
+		instance: instance,
+	}
+	if err := module.prep(); err != nil {
+		return nil, err
+	}
+	return module, nil
+}
+
+func (c *Control) Manager() *mgr.Manager {
+	return c.mgr
+}
+
+func (c *Control) Start() error {
+	return nil
+}
+
+func (c *Control) Stop() error {
+	c.locker.Lock()
+	defer c.locker.Unlock()
+	c.stopResumeWorker()
+
+	return nil
+}
+
+func (c *Control) prep() error {
+	return c.registerAPIEndpoints()
+}

service/control/pause.go

@@ -0,0 +1,192 @@
+package control
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/safing/portmaster/base/api"
+	"github.com/safing/portmaster/base/config"
+	"github.com/safing/portmaster/service/mgr"
+)
+
+func (c *Control) handlePause(r *api.Request) (msg string, err error) {
+	params := pauseRequestParams{}
+	if r.InputData != nil {
+		if err := json.Unmarshal(r.InputData, &params); err != nil {
+			return "Bad Request: invalid input data", err
+		}
+	}
+
+	if params.OnlySPN {
+		c.mgr.Info(fmt.Sprintf("Received SPN PAUSE(%v) action request ", params.Duration))
+	} else {
+		c.mgr.Info(fmt.Sprintf("Received PAUSE(%v) action request ", params.Duration))
+	}
+
+	if err := c.impl_pause(time.Duration(params.Duration)*time.Second, params.OnlySPN); err != nil {
+		return "Failed to pause", err
+	}
+	return "Pause initiated", nil
+}
+
+func (c *Control) handleResume(_ *api.Request) (msg string, err error) {
+	c.mgr.Info("Received RESUME action request")
+	if err := c.impl_resume(); err != nil {
+		return "Failed to resume", err
+	}
+	return "Resume initiated", nil
+}
+
+func (c *Control) impl_pause(duration time.Duration, onlySPN bool) (retErr error) {
+	c.locker.Lock()
+	defer c.locker.Unlock()
+
+	if duration <= 0 {
+		return errors.New(logPrefix + "invalid pause duration")
+	}
+
+	if onlySPN {
+		if c.isPaused {
+			return errors.New(logPrefix + "cannot pause SPN separately when core is paused")
+		}
+		if !c.isPausedSPN && !c.instance.SPNGroup().Ready() {
+			return errors.New(logPrefix + "cannot pause SPN when it is not running")
+		}
+	}
+
+	c.stopResumeWorker()
+	defer func() {
+		if retErr == nil {
+			// start new resume worker (with new duration) if no error
+			c.startResumeWorker(duration)
+		}
+	}()
+
+	if !c.isPausedSPN {
+		if c.instance.SPNGroup().Ready() {
+
+			// TODO: the 'pause' state must not make permanent config changes.
+			// Consider possibility to not store permanent config changes.
+			// E.g. SPN enabled -> pause SPN -> restart PC/Portmaster -> SPN should be enabled again.
+			enabled := config.GetAsBool("spn/enable", false)
+			if enabled() {
+				config.SetConfigOption("spn/enable", false)
+			}
+
+			// Alternatively, we could directly stop SPN here:
+			//  if c.instance.IsShuttingDown() {
+			//	  c.mgr.Warn("Skipping pause during shutdown")
+			//	  return nil
+			//  }
+			//	err := c.instance.SPNGroup().Stop()
+			//	if err != nil {
+			//		return err
+			//	}
+			//	c.mgr.Info("SPN paused")
+
+			c.isPausedSPN = true
+		}
+	}
+
+	if onlySPN {
+		return nil
+	}
+	if c.isPaused {
+		return nil
+	}
+
+	modulesToResume := []mgr.Module{
+		c.instance.Compat(),
+		c.instance.Interception(),
+	}
+	for _, m := range modulesToResume {
+		if err := m.Stop(); err != nil {
+			return err
+		}
+	}
+
+	c.mgr.Info("interception paused")
+	c.isPaused = true
+
+	return nil
+}
+
+func (c *Control) impl_resume() error {
+	c.locker.Lock()
+	defer c.locker.Unlock()
+
+	c.stopResumeWorker()
+
+	if c.isPausedSPN {
+
+		// TODO: consider using event to handle  "spn/enable" changes:
+		//	 	module.instance.Config().EventConfigChange
+		enabled := config.GetAsBool("spn/enable", false)
+		if !enabled() {
+			config.SetConfigOption("spn/enable", true)
+		}
+
+		// Alternatively, we could directly start SPN here:
+		//  if c.instance.IsShuttingDown() {
+		// 	 c.mgr.Warn("Skipping resume during shutdown")
+		// 	 return nil
+		//  }
+		//	if !c.instance.SPNGroup().Ready() {
+		//		err := c.instance.SPNGroup().Start()
+		//		if err != nil {
+		//			return err
+		//		}
+		//		c.mgr.Info("SPN resumed")
+		//	}
+
+		c.isPausedSPN = false
+	}
+
+	if c.isPaused {
+		modulesToResume := []mgr.Module{
+			c.instance.Interception(),
+			c.instance.Compat(),
+		}
+
+		for _, m := range modulesToResume {
+			if err := m.Start(); err != nil {
+				return err
+			}
+		}
+		c.mgr.Info("interception resumed")
+		c.isPaused = false
+	}
+
+	return nil
+}
+
+// stopResumeWorker stops any existing resume worker.
+// No thread safety, caller must hold c.locker.
+func (c *Control) stopResumeWorker() {
+	c.pauseStartTime = time.Time{}
+	c.pauseDuration = 0
+
+	if c.pauseWorker != nil {
+		c.pauseWorker.Stop()
+		c.pauseWorker = nil
+	}
+}
+
+// startResumeWorker starts a worker that will resume normal operation after the specified duration.
+// No thread safety, caller must hold c.locker.
+func (c *Control) startResumeWorker(duration time.Duration) {
+	c.pauseStartTime = time.Now()
+	c.pauseDuration = duration
+
+	c.mgr.Info(fmt.Sprintf("Scheduling resume in %v", duration))
+
+	c.pauseWorker = c.mgr.NewWorkerMgr(
+		fmt.Sprintf("resume in %v", duration),
+		func(wc *mgr.WorkerCtx) error {
+			wc.Info("Resuming...")
+			return c.impl_resume()
+		},
+		nil).Delay(duration)
+}

service/firewall/interception/interception_linux.go

@@ -34,6 +34,11 @@ func startInterception(packets chan packet.Packet) error {
 
 // stop starts the interception.
 func stopInterception() error {
+	// TODO: stop ebpf workers gracefully
+	// E.g.:
+	// 		module.mgr.Cancel()
+	// 		<-m.Done()
+
 	return StopNfqueueInterception()
 }
 

service/firewall/interception/introspection.go

@@ -12,9 +12,7 @@ import (
 
 var (
 	packetMetricsDestination string
-	metrics                  = &packetMetrics{
-		done: make(chan struct{}),
-	}
+	metrics                  = &packetMetrics{}
 )
 
 func init() {
@@ -40,6 +38,10 @@ func (pm *packetMetrics) record(tp *tracedPacket, verdict string) {
 		pm.l.Lock()
 		defer pm.l.Unlock()
 
+		if pm.done == nil {
+			return
+		}
+
 		pm.records = append(pm.records, &performanceRecord{
 			start:    start,
 			duration: duration,
@@ -48,6 +50,18 @@ func (pm *packetMetrics) record(tp *tracedPacket, verdict string) {
 	}(tp.start.UnixNano(), time.Since(tp.start))
 }
 
+func (pm *packetMetrics) stop() {
+	pm.l.Lock()
+	defer pm.l.Unlock()
+
+	if pm.done == nil {
+		return
+	}
+
+	close(pm.done)
+	pm.done = nil
+}
+
 func (pm *packetMetrics) writeMetrics() {
 	if packetMetricsDestination == "" {
 		return
@@ -62,9 +76,14 @@ func (pm *packetMetrics) writeMetrics() {
 		_ = f.Close()
 	}()
 
+	pm.l.Lock()
+	pm.done = make(chan struct{})
+	done := pm.done
+	pm.l.Unlock()
+
 	for {
 		select {
-		case <-pm.done:
+		case <-done:
 			return
 		case <-time.After(time.Second * 5):
 		}

service/firewall/interception/module.go

@@ -40,6 +40,7 @@ var (
 	BandwidthUpdates = make(chan *packet.BandwidthUpdate, 1000)
 
 	disableInterception bool
+	isStarted           atomic.Bool
 )
 
 func init() {
@@ -53,6 +54,10 @@ func start() error {
 		return nil
 	}
 
+	if !isStarted.CompareAndSwap(false, true) {
+		return nil // already running
+	}
+
 	inputPackets := Packets
 	if packetMetricsDestination != "" {
 		go metrics.writeMetrics()
@@ -64,7 +69,16 @@ func start() error {
 		}()
 	}
 
-	return startInterception(inputPackets)
+	err := startInterception(inputPackets)
+	if err != nil {
+		log.Errorf("interception: failed to start module: %q", err)
+		log.Debug("interception: cleaning up after failed start...")
+		if e := stopInterception(); e != nil {
+			log.Debugf("interception: error cleaning up after failed start: %q", e.Error())
+		}
+		isStarted.Store(false)
+	}
+	return err
 }
 
 // Stop starts the interception.
@@ -73,7 +87,11 @@ func stop() error {
 		return nil
 	}
 
-	close(metrics.done)
+	if !isStarted.CompareAndSwap(true, false) {
+		return nil // not running
+	}
+
+	metrics.stop()
 	if err := stopInterception(); err != nil {
 		log.Errorf("failed to stop interception module: %s", err)
 	}

service/instance.go

@@ -16,6 +16,7 @@ import (
 	"github.com/safing/portmaster/base/utils"
 	"github.com/safing/portmaster/service/broadcasts"
 	"github.com/safing/portmaster/service/compat"
+	"github.com/safing/portmaster/service/control"
 	"github.com/safing/portmaster/service/core"
 	"github.com/safing/portmaster/service/core/base"
 	"github.com/safing/portmaster/service/firewall"
@@ -95,6 +96,7 @@ type Instance struct {
 	process       *process.ProcessModule
 	resolver      *resolver.ResolverModule
 	sync          *sync.Sync
+	control       *control.Control
 
 	access *access.Access
 
@@ -264,6 +266,10 @@ func New(svcCfg *ServiceConfig) (*Instance, error) { //nolint:maintidx
 	if err != nil {
 		return instance, fmt.Errorf("create sync module: %w", err)
 	}
+	instance.control, err = control.New(instance)
+	if err != nil {
+		return instance, fmt.Errorf("create control module: %w", err)
+	}
 	instance.access, err = access.New(instance)
 	if err != nil {
 		return instance, fmt.Errorf("create access module: %w", err)
@@ -342,6 +348,7 @@ func New(svcCfg *ServiceConfig) (*Instance, error) { //nolint:maintidx
 		instance.broadcasts,
 		instance.sync,
 		instance.ui,
+		instance.control,
 
 		instance.access,
 	)