[CWS] fix race between active ruleset and replay (#45088)

### What does this PR do?

This PR ensures that the replay events are evaluation against the current ruleset loaded. Before the replay events were potentially evaluated against the previous ruleset. 

### Motivation

### Describe how you validated your changes

### Additional Notes


Co-authored-by: sylvain.afchain <sylvain.afchain@datadoghq.com>

Author: safchain

pkg/security/probe/probe.go

@@ -48,7 +48,7 @@ type PlatformProbe interface {
 	NewModel() *model.Model
 	DumpDiscarders() (string, error)
 	FlushDiscarders() error
-	ApplyRuleSet(_ *rules.RuleSet) (*kfilters.FilterReport, error)
+	ApplyRuleSet(_ *rules.RuleSet) (*kfilters.FilterReport, bool, error)
 	OnNewRuleSetLoaded(_ *rules.RuleSet)
 	OnNewDiscarder(_ *rules.RuleSet, _ *model.Event, _ eval.Field, _ eval.EventType)
 	HandleActions(_ *eval.Context, _ *rules.Rule)
@@ -58,6 +58,7 @@ type PlatformProbe interface {
 	AddDiscarderPushedCallback(_ DiscarderPushedCallback)
 	GetEventTags(_ containerutils.ContainerID) []string
 	EnableEnforcement(bool)
+	ReplayEvents()
 }
 
 var probeTelemetry = struct {
@@ -221,7 +222,7 @@ func (p *Probe) FlushDiscarders() error {
 }
 
 // ApplyRuleSet setup the probes for the provided set of rules and returns the policy report.
-func (p *Probe) ApplyRuleSet(rs *rules.RuleSet) (*kfilters.FilterReport, error) {
+func (p *Probe) ApplyRuleSet(rs *rules.RuleSet) (*kfilters.FilterReport, bool, error) {
 	return p.PlatformProbe.ApplyRuleSet(rs)
 }
 
@@ -409,6 +410,11 @@ func (p *Probe) onRuleActionPerformed(rule *rules.Rule, action *rules.ActionDefi
 	}
 }
 
+// ReplayEvents replays the events from the rule set
+func (p *Probe) ReplayEvents() {
+	p.PlatformProbe.ReplayEvents()
+}
+
 // NewRuleSet returns a new ruleset
 func (p *Probe) NewRuleSet(eventTypeEnabled map[eval.EventType]bool) *rules.RuleSet {
 	ruleOpts, evalOpts := rules.NewBothOpts(eventTypeEnabled)

pkg/security/probe/probe_ebpf.go

@@ -802,6 +802,16 @@ func (p *EBPFProbe) Start() error {
 	return p.eventStream.Start(&p.wg)
 }
 
+// ReplayEvents replays the events from the rule set
+func (p *EBPFProbe) ReplayEvents() {
+	p.replayEventsState.Store(true)
+
+	// trigger a nop event to ensure that the replayed events are processed
+	if err := p.triggerNopEvent(); err != nil {
+		seclog.Warnf("unable to trigger nop event: %v", err)
+	}
+}
+
 func (p *EBPFProbe) replayEvents(notifyConsumers bool) {
 	seclog.Debugf("playing the snapshot")
 
@@ -2416,16 +2426,16 @@ func isRawPacketActionPresent(rs *rules.RuleSet) bool {
 }
 
 // ApplyRuleSet apply the required update to handle the new ruleset
-func (p *EBPFProbe) ApplyRuleSet(rs *rules.RuleSet) (*kfilters.FilterReport, error) {
+func (p *EBPFProbe) ApplyRuleSet(rs *rules.RuleSet) (*kfilters.FilterReport, bool, error) {
 	if p.opts.SyscallsMonitorEnabled {
 		if err := p.monitors.syscallsMonitor.Disable(); err != nil {
-			return nil, err
+			return nil, false, err
 		}
 	}
 
 	filterReport, err := kfilters.ComputeFilters(p.config.Probe, rs)
 	if err != nil {
-		return nil, err
+		return nil, false, err
 	}
 
 	if err := applyDNSDefaultDropMaskFromRules(p.Manager, rs); err != nil {
@@ -2438,16 +2448,16 @@ func (p *EBPFProbe) ApplyRuleSet(rs *rules.RuleSet) (*kfilters.FilterReport, err
 		if err := p.setApprovers(eventType, report.Approvers); err != nil {
 			seclog.Errorf("Error while adding approvers fallback in-kernel policy to `%s` for `%s`: %s", kfilters.PolicyModeAccept, eventType, err)
 			if err := fpb.add(eventType, kfilters.PolicyModeAccept); err != nil {
-				return nil, err
+				return nil, false, err
 			}
 		} else {
 			if err := fpb.add(eventType, report.Mode); err != nil {
-				return nil, err
+				return nil, false, err
 			}
 		}
 	}
 	if err := fpb.apply(p.Manager); err != nil {
-		return nil, fmt.Errorf("unable to apply to filter policy: %w", err)
+		return nil, false, fmt.Errorf("unable to apply to filter policy: %w", err)
 	}
 
 	eventTypes := rs.GetEventTypes()
@@ -2488,15 +2498,15 @@ func (p *EBPFProbe) ApplyRuleSet(rs *rules.RuleSet) (*kfilters.FilterReport, err
 	}
 
 	if err := p.updateProbes(eventTypes, needRawSyscalls); err != nil {
-		return nil, fmt.Errorf("failed to select probes: %w", err)
+		return nil, false, fmt.Errorf("failed to select probes: %w", err)
 	}
 
 	if p.opts.SyscallsMonitorEnabled {
 		if err := p.monitors.syscallsMonitor.Flush(); err != nil {
-			return nil, err
+			return nil, false, err
 		}
 		if err := p.monitors.syscallsMonitor.Enable(); err != nil {
-			return nil, err
+			return nil, false, err
 		}
 	}
 
@@ -2520,18 +2530,10 @@ func (p *EBPFProbe) ApplyRuleSet(rs *rules.RuleSet) (*kfilters.FilterReport, err
 	}
 
 	// do not replay the snapshot if we are in the first rule set version, this was already done in the start method
-	if p.ruleSetVersion != 0 {
-		p.replayEventsState.Store(true)
-	}
-
+	replayEvents := p.ruleSetVersion != 0
 	p.ruleSetVersion++
 
-	// trigger a nop event to ensure that the replayed events are processed
-	if err := p.triggerNopEvent(); err != nil {
-		seclog.Warnf("unable to trigger nop event: %v", err)
-	}
-
-	return filterReport, nil
+	return filterReport, replayEvents, nil
 }
 
 // OnNewRuleSetLoaded resets statistics and states once a new rule set is loaded

pkg/security/probe/probe_ebpfless.go

@@ -636,8 +636,12 @@ func (p *EBPFLessProbe) FlushDiscarders() error {
 }
 
 // ApplyRuleSet applies the new ruleset
-func (p *EBPFLessProbe) ApplyRuleSet(_ *rules.RuleSet) (*kfilters.FilterReport, error) {
-	return &kfilters.FilterReport{}, nil
+func (p *EBPFLessProbe) ApplyRuleSet(_ *rules.RuleSet) (*kfilters.FilterReport, bool, error) {
+	return &kfilters.FilterReport{}, false, nil
+}
+
+// ReplayEvents replays the events from the rule set
+func (p *EBPFLessProbe) ReplayEvents() {
 }
 
 // OnNewRuleSetLoaded resets statistics and states once a new rule set is loaded

pkg/security/probe/probe_others.go

@@ -53,8 +53,8 @@ func (p *Probe) NewRuleSet(_ map[eval.EventType]bool) *rules.RuleSet {
 }
 
 // ApplyRuleSet setup the probes for the provided set of rules and returns the policy report.
-func (p *Probe) ApplyRuleSet(_ *rules.RuleSet) (*kfilters.FilterReport, error) {
-	return nil, nil
+func (p *Probe) ApplyRuleSet(_ *rules.RuleSet) (*kfilters.FilterReport, bool, error) {
+	return nil, false, nil
 }
 
 // OnNewRuleSetLoaded resets statistics and states once a new rule set is loaded
@@ -85,6 +85,10 @@ func (p *Probe) IsNetworkEnabled() bool {
 	return p.Config.Probe.NetworkEnabled
 }
 
+// ReplayEvents replays the events from the rule set
+func (p *Probe) ReplayEvents() {
+}
+
 // IsNetworkRawPacketEnabled returns whether network raw packet is enabled
 func (p *Probe) IsNetworkRawPacketEnabled() bool {
 	return p.IsNetworkEnabled() && p.Config.Probe.NetworkRawPacketEnabled

pkg/security/probe/probe_windows.go

@@ -1169,6 +1169,10 @@ func (p *WindowsProbe) setProcessContext(pid uint32, event *model.Event) error {
 	return err
 }
 
+// ReplayEvents replays the events from the rule set
+func (p *WindowsProbe) ReplayEvents() {
+}
+
 // DispatchEvent sends an event to the probe event handler
 func (p *WindowsProbe) DispatchEvent(event *model.Event) {
 	p.probe.logTraceEvent(event.GetEventType(), event)
@@ -1446,7 +1450,7 @@ func NewWindowsProbe(probe *Probe, config *config.Config, ipc ipc.Component, opt
 }
 
 // ApplyRuleSet setup the probes for the provided set of rules and returns the policy report.
-func (p *WindowsProbe) ApplyRuleSet(rs *rules.RuleSet) (*kfilters.FilterReport, error) {
+func (p *WindowsProbe) ApplyRuleSet(rs *rules.RuleSet) (*kfilters.FilterReport, bool, error) {
 	p.enabledEventTypesLock.Lock()
 	clear(p.enabledEventTypes)
 	for _, eventType := range rs.GetEventTypes() {
@@ -1456,7 +1460,7 @@ func (p *WindowsProbe) ApplyRuleSet(rs *rules.RuleSet) (*kfilters.FilterReport,
 
 	filterReport, err := kfilters.ComputeFilters(p.config.Probe, rs)
 	if err != nil {
-		return nil, err
+		return nil, false, err
 	}
 
 	// remove old approvers
@@ -1466,17 +1470,17 @@ func (p *WindowsProbe) ApplyRuleSet(rs *rules.RuleSet) (*kfilters.FilterReport,
 
 	for eventType, report := range filterReport.ApproverReports {
 		if err := p.setApprovers(eventType, report.Approvers); err != nil {
-			return nil, err
+			return nil, false, err
 		}
 	}
 
 	p.enabledEventTypesLock.RLock()
 	defer p.enabledEventTypesLock.RUnlock()
 	if err := p.reconfigureProvider(); err != nil {
-		return nil, err
+		return nil, false, err
 	}
 
-	return filterReport, nil
+	return filterReport, false, nil
 }
 
 // OnNewRuleSetLoaded resets statistics and states once a new rule set is loaded

pkg/security/rules/engine.go

@@ -358,7 +358,7 @@ func (e *RuleEngine) LoadPolicies(providers []rules.PolicyProvider, sendLoadedRe
 	ruleIDs = append(ruleIDs, events.AllCustomRuleIDs()...)
 
 	// analyze the ruleset, push probe evaluation rule sets to the kernel and generate the policy report
-	filterReport, err := e.probe.ApplyRuleSet(rs)
+	filterReport, replayEvents, err := e.probe.ApplyRuleSet(rs)
 	if err != nil {
 		return err
 	}
@@ -371,6 +371,10 @@ func (e *RuleEngine) LoadPolicies(providers []rules.PolicyProvider, sendLoadedRe
 
 	e.currentRuleSet.Store(rs)
 
+	if replayEvents {
+		e.probe.ReplayEvents()
+	}
+
 	if err := e.probe.FlushDiscarders(); err != nil {
 		return fmt.Errorf("failed to flush discarders: %w", err)
 	}