feat(remediation): free-core execution engine (Fix + rollback, kensa v0.5.1)

audit/events.yaml

@@ -952,6 +952,31 @@ events:
   - code: admin.user.deleted
     severity: warning
 
+  - code: admin.user.password_reset
+    severity: warning
+    description: An administrator reset another user's (or their own) password.
+    detail_schema:
+      type: object
+      properties:
+        target_user_id: {type: string}
+        self: {type: boolean}
+
+  - code: admin.user.disabled
+    severity: warning
+    description: An administrator disabled a user account (cannot authenticate).
+    detail_schema:
+      type: object
+      properties:
+        target_user_id: {type: string}
+
+  - code: admin.user.enabled
+    severity: warning
+    description: An administrator re-enabled a previously disabled user account.
+    detail_schema:
+      type: object
+      properties:
+        target_user_id: {type: string}
+
   - code: admin.role.changed
     severity: warning
 

auth/permissions.yaml

@@ -313,15 +313,13 @@ permissions:
 
   - id: remediation:execute
     category: remediation
-    description: Execute an approved remediation against hosts
+    description: Execute an approved single-rule remediation against a host (free core)
     dangerous: true
-    license_gated: remediation_execution
 
   - id: remediation:rollback
     category: remediation
-    description: Roll back a previously executed remediation
+    description: Roll back a previously executed remediation (free core)
     dangerous: true
-    license_gated: remediation_execution
 
   # =========================================================================
   # integration - plugins and webhooks
@@ -511,6 +509,8 @@ roles:
       - policy:read
       - remediation:read
       - remediation:request
+      - remediation:execute
+      - remediation:rollback
       - integration:read
       - audit:read
       - system:read

cmd/openwatch/main.go

@@ -17,6 +17,7 @@ import (
 	"log/slog"
 	"os"
 	"os/signal"
+	"path/filepath"
 	"runtime"
 	"syscall"
 	"time"
@@ -50,6 +51,7 @@ import (
 	openlog "github.com/Hanalyx/openwatch/internal/log"
 	"github.com/Hanalyx/openwatch/internal/notification"
 	"github.com/Hanalyx/openwatch/internal/posture"
+	"github.com/Hanalyx/openwatch/internal/remediation"
 	"github.com/Hanalyx/openwatch/internal/report"
 	"github.com/Hanalyx/openwatch/internal/scanresult"
 	compsched "github.com/Hanalyx/openwatch/internal/scheduler"
@@ -573,10 +575,54 @@ func cmdServe(cfg *config.Config, _ []string, stdout, stderr *os.File) int {
 	exceptionSvc := exception.NewService(pool, audit.Emit)
 	exceptionSvc.Run(ctx, 0)
 
+	// Remediation governance: request/approve/reject + projected lift (free
+	// core), AND the queued single-rule execute/rollback (Tier A free core).
+	// Spec api-remediation.
+	remediationSvc := remediation.NewService(pool, audit.Emit)
+	remTxWriter := transactionlog.NewWriter(pool, audit.Emit)
+
+	// Remediation execution executor: shares the scan executor's per-host
+	// inFlight guard by chaining WithRemediateFunc onto it (so a host is never
+	// scanned + remediated at the same instant). The apply-enabled Kensa needs
+	// a durable SQLite store for rollback pre-state — derive a path from the
+	// kensa store env (dev default under the working dir).
+	remExecutor := scanExecutor
+	if remFn, rbFn, remErr := kensa.NewProductionRemediateFunc(bootCtx, kensa.RemediateFuncDeps{
+		Pool:        pool,
+		Credentials: credSvc,
+		RulesDir:    scanRulesDir,
+		HostKeyMode: owssh.ModeTOFU,
+		KnownHosts:  knownhosts.NewStore(pool),
+		Variables: func(ctx context.Context) (map[string]string, error) {
+			vars, err := cfgStore.LoadScanVars(ctx)
+			return vars, err
+		},
+		Profiles: connStore,
+		Policy: func(ctx context.Context) (bool, error) {
+			cfg, err := cfgStore.LoadSecurity(ctx)
+			return cfg.AllowCredentialSudoPassword, err
+		},
+		StorePath: kensaStorePath(bootCtx),
+	}); remErr != nil {
+		slog.WarnContext(bootCtx, "kensa remediation wiring unavailable — remediation execute/rollback will fail until the kensa-rules package is installed (or OPENWATCH_KENSA_RULES_DIR set)",
+			slog.String("error", remErr.Error()))
+	} else {
+		remExecutor = remExecutor.WithRemediateFunc(remFn, rbFn)
+	}
+	remediationWorker := worker.NewRemediationWorker(worker.RemediationConfig{
+		Pool:     pool,
+		Executor: remExecutor,
+		Service:  remediationSvc,
+		Writer:   remTxWriter,
+		QueueKey: scanQueueKey,
+		Bus:      bus,
+		Emit:     audit.Emit,
+	})
+
 	scanWorker := worker.NewScanWorker(worker.Config{
 		Pool:        pool,
 		Executor:    scanExecutor,
-		Writer:      transactionlog.NewWriter(pool, audit.Emit),
+		Writer:      remTxWriter,
 		ScanResults: scanresult.NewWriter(pool),
 		QueueKey:    scanQueueKey,
 		Emit:        audit.Emit,
@@ -592,10 +638,12 @@ func cmdServe(cfg *config.Config, _ []string, stdout, stderr *os.File) int {
 		WithAlerts(alerts.NewService(pool, audit.Emit)).
 		WithScanQueue(scanQueueKey).
 		WithScanWorker(scanWorker).
+		WithRemediationWorker(remediationWorker).
 		WithRuleCatalog(ruleCatalog).
 		WithRuleLibrary(ruleLibrary).
 		WithVariableCatalog(varCatalog).
 		WithExceptions(exceptionSvc).
+		WithRemediation(remediationSvc).
 		WithGroups(group.NewService(pool)).
 		WithReports(report.NewService(pool)).
 		WithScanResults(scanresult.NewReader(pool)).
@@ -653,6 +701,26 @@ func (a collectorSSHAdapter) Dial(ctx context.Context, host string, port int, cr
 	return sess, nil
 }
 
+// kensaStorePath resolves the durable SQLite path Kensa uses for remediation
+// rollback pre-state. Resolution order:
+//
+//	OPENWATCH_KENSA_STORE_PATH   explicit override (production: a durable path
+//	                             under the data dir, e.g.
+//	                             /var/lib/openwatch/kensa/remediation.db)
+//	<workdir>/.kensa/remediation.db   dev default (warned)
+//
+// The pre-state log MUST survive restarts for rollback to work, so production
+// installs set the env to a persistent location.
+func kensaStorePath(ctx context.Context) string {
+	if p := os.Getenv("OPENWATCH_KENSA_STORE_PATH"); p != "" {
+		return p
+	}
+	def := filepath.Join(".kensa", "remediation.db")
+	slog.WarnContext(ctx, "OPENWATCH_KENSA_STORE_PATH unset — using working-dir default for kensa rollback pre-state; production must set a durable path",
+		slog.String("store_path", def))
+	return def
+}
+
 // parseLogLevel maps the config string to a slog.Level. Unknown values
 // default to info (Validate would have caught them earlier).
 func parseLogLevel(s string) slog.Level {

cmd/openwatch/worker.go

@@ -30,6 +30,7 @@ import (
 	"github.com/Hanalyx/openwatch/internal/knownhosts"
 	"github.com/Hanalyx/openwatch/internal/license"
 	openlog "github.com/Hanalyx/openwatch/internal/log"
+	"github.com/Hanalyx/openwatch/internal/remediation"
 	"github.com/Hanalyx/openwatch/internal/scanresult"
 	"github.com/Hanalyx/openwatch/internal/scheduler"
 	"github.com/Hanalyx/openwatch/internal/secretkey"
@@ -200,6 +201,45 @@ func cmdWorker(cfg *config.Config, args []string, stdout, stderr *os.File) int {
 	writer := transactionlog.NewWriter(pool, audit.Emit)
 	scanResultsWriter := scanresult.NewWriter(pool)
 
+	// Remediation execution wiring (Tier A free core): chain the apply-enabled
+	// Remediate/Rollback seams onto the same executor so a host's scan +
+	// remediate share one per-host inFlight guard. The apply-enabled Kensa
+	// needs a durable SQLite store for rollback pre-state.
+	remExecutor := executor
+	remFn, rbFn, remErr := kensa.NewProductionRemediateFunc(bootCtx, kensa.RemediateFuncDeps{
+		Pool:        pool,
+		Credentials: credSvc,
+		RulesDir:    rulesDir,
+		HostKeyMode: owssh.ModeTOFU,
+		KnownHosts:  knownhosts.NewStore(pool),
+		Variables: func(ctx context.Context) (map[string]string, error) {
+			vars, err := varStore.LoadScanVars(ctx)
+			return vars, err
+		},
+		Profiles: connprofile.NewStore(pool),
+		Policy: func(ctx context.Context) (bool, error) {
+			cfg, err := varStore.LoadSecurity(ctx)
+			return cfg.AllowCredentialSudoPassword, err
+		},
+		StorePath: kensaStorePath(bootCtx),
+	})
+	if remErr != nil {
+		slog.WarnContext(bootCtx, "kensa remediation wiring unavailable — remediation jobs claimed by this worker will fail",
+			slog.String("error", remErr.Error()))
+	} else {
+		remExecutor = remExecutor.WithRemediateFunc(remFn, rbFn)
+	}
+	remediationWorker := worker.NewRemediationWorker(worker.RemediationConfig{
+		Pool:     pool,
+		Executor: remExecutor,
+		Service:  remediation.NewService(pool, audit.Emit),
+		Writer:   writer,
+		QueueKey: queueKey,
+		Emit:     audit.Emit,
+		// Bus nil: the dedicated worker has no SSE subscribers (cross-process
+		// delivery is a known non-goal, same as scan.completed).
+	})
+
 	// Post-scan schedule updates run here too: the dedicated worker
 	// classifies each completed scan into a compliance state so
 	// host_compliance_schedule stays fresh whichever process executed
@@ -217,14 +257,15 @@ func cmdWorker(cfg *config.Config, args []string, stdout, stderr *os.File) int {
 		!scanCfg.Enabled || scanCfg.MaintenanceGlobal)
 
 	scanWorker := worker.NewScanWorker(worker.Config{
-		Pool:         pool,
-		Executor:     executor,
-		Writer:       writer,
-		ScanResults:  scanResultsWriter,
-		QueueKey:     queueKey,
-		PollInterval: *pollInterval,
-		Emit:         audit.Emit,
-		Sched:        sched,
+		Pool:                 pool,
+		Executor:             executor,
+		Writer:               writer,
+		ScanResults:          scanResultsWriter,
+		QueueKey:             queueKey,
+		PollInterval:         *pollInterval,
+		Emit:                 audit.Emit,
+		Sched:                sched,
+		RemediationProcessor: remediationWorker,
 	})
 
 	ctx, stop := signal.NotifyContext(bootCtx, syscall.SIGINT, syscall.SIGTERM)