feat(watcher): add fanotify-based file access tracking for linux

.gitignore

@@ -1,2 +1,4 @@
 gorm.db
 cover.out
+/.idea/
+/bin/

Makefile

@@ -324,3 +324,10 @@ mv $(1) $(1)-$(3) ;\
 } ;\
 ln -sf $(1)-$(3) $(1)
 endef
+## Lint Go code with golangci-lint
+.PHONY: lint lint-fix
+lint:
+	golangci-lint run --timeout=5m
+
+lint-fix:
+	golangci-lint run --fix --timeout=5m

WATCHER.md

@@ -0,0 +1,52 @@
+Watcher integration (Linux only)
+
+Overview
+- vm_server now starts a fanotify-based watcher (when enabled) and records "consumers" — processes that access files under the requested scan paths.
+- Consumers are upserted with FilePath, Comm, Exe, RUID, EUID and timestamps (first/last seen).
+
+Flags
+- `-watch-enable` (default: true): enable/disable watcher.
+- `-watch-mount` (default: false): watch the whole mount for each provided path (more coverage, noisier). When false, a directory mark is used (non-recursive; child events only).
+- `-watch-verbose` (default: false): verbose watcher logs.
+
+API
+- Start a scan: `POST /api/v1/scan` with body `{ "paths": ["/abs/dir"], "regexes": [], "threshold": 1 }`
+- List consumers: `GET /api/v1/consumers?filePath=/abs/file/path`
+
+Ubuntu VM quick start
+1) Build vm_server (Linux):
+   - `cd vm_server`
+   - `go mod tidy`
+   - `GOOS=linux GOARCH=$(uname -m | sed 's/x86_64/amd64/;s/aarch64/arm64/') go build -o eso-vm-server .`
+2) Copy to VM and run as root (fanotify needs privileges):
+   - `sudo ./eso-vm-server -port=1323 -watch-enable=true -watch-verbose=true`
+3) Start a scan against a directory:
+   - `curl -s -X POST http://localhost:1323/api/v1/scan -H 'content-type: application/json' -d '{"paths":["/home/ubuntu/watch"],"regexes":[],"threshold":1}'`
+4) Generate activity under the path (new shell):
+   - `echo hi >> /home/ubuntu/watch/file && cat /home/ubuntu/watch/file`
+5) Query consumers:
+   - `curl -s 'http://localhost:1323/api/v1/consumers?filePath=/home/ubuntu/watch/file' | jq`
+
+Notes
+- Directory marks are not recursive; events fire for direct children. Use `-watch-mount` for broad coverage.
+- Running inside containers may require `--privileged`, `--pid=host`, and host bind mounts to see host activity.
+- On non-Linux platforms, the watcher is disabled at build time; the server still runs without consumer tracking.
+
+Permissions and capabilities
+- Simplest: run as root (sudo). Fanotify and /proc reads work reliably.
+- Non-root options:
+  - Grant capabilities to the binary or via systemd:
+    - `CAP_SYS_ADMIN` (required for mount marks; generally needed for fanotify in practice)
+    - `CAP_DAC_READ_SEARCH` (improves ability to read `/proc/<pid>` for comm/exe/uid)
+    - Example: `sudo setcap 'cap_sys_admin,cap_dac_read_search+ep' /usr/local/bin/eso-vm-server`
+  - Ensure seccomp/AppArmor do not block fanotify syscalls (use unconfined profile if needed).
+  - `/proc` may be mounted with `hidepid=2` on hardened systems; this blocks reading other PIDs. Options:
+    - Run as root, or
+    - Remount `/proc` with `hidepid=0` (security trade-off), or
+    - Add `CAP_DAC_READ_SEARCH`.
+  - Containers/K8s: use `--cap-add SYS_ADMIN`, `--pid=host`, and unconfined seccomp/AppArmor; or `--privileged` for PoC.
+
+Startup checks
+- On startup, the watcher emits warnings if:
+  - `-watch-mount` is set but CAP_SYS_ADMIN is missing.
+  - `/proc/<pid>/status` cannot be read for other processes (likely hidepid/permissions). It suggests fixes.

go.mod

@@ -3,12 +3,15 @@ module vm-server
 go 1.24.2
 
 require (
+	github.com/google/uuid v1.6.0
 	github.com/labstack/echo/v4 v4.13.4
+	github.com/s3rj1k/go-fanotify/fanotify v0.0.0-20240229202106-bca3154da60a
+	golang.org/x/sys v0.33.0
+	gorm.io/driver/sqlite v1.6.0
 	gorm.io/gorm v1.30.0
 )
 
 require (
-	github.com/google/uuid v1.6.0 // indirect
 	github.com/jinzhu/inflection v1.0.0 // indirect
 	github.com/jinzhu/now v1.1.5 // indirect
 	github.com/labstack/gommon v0.4.2 // indirect
@@ -19,8 +22,6 @@ require (
 	github.com/valyala/fasttemplate v1.2.2 // indirect
 	golang.org/x/crypto v0.38.0 // indirect
 	golang.org/x/net v0.40.0 // indirect
-	golang.org/x/sys v0.33.0 // indirect
 	golang.org/x/text v0.25.0 // indirect
 	golang.org/x/time v0.11.0 // indirect
-	gorm.io/driver/sqlite v1.6.0 // indirect
 )

go.sum

@@ -18,6 +18,8 @@ github.com/mattn/go-sqlite3 v1.14.22 h1:2gZY6PC6kBnID23Tichd1K+Z0oS6nE/XwU+Vz/5o
 github.com/mattn/go-sqlite3 v1.14.22/go.mod h1:Uh1q+B4BYcTPb+yiD3kU8Ct7aC0hY9fxUwlHK0RXw+Y=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/s3rj1k/go-fanotify/fanotify v0.0.0-20240229202106-bca3154da60a h1:4VFls9SuqkqeioVevnaeTXrYKQ7JiEsxqKHfxp+/ovA=
+github.com/s3rj1k/go-fanotify/fanotify v0.0.0-20240229202106-bca3154da60a/go.mod h1:2zG1g57bc+D6FpNc68gsRXJgkidteqTMhWiiUP3m8UE=
 github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA=
 github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
 github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw=

handlers/consumers/consumers.go

@@ -0,0 +1,25 @@
+package consumers
+
+import (
+	"net/http"
+	"vm-server/models"
+	services "vm-server/services/scan"
+
+	"github.com/labstack/echo/v4"
+)
+
+type Handler struct {
+	Service *services.Service
+}
+
+func NewHandler(s *services.Service) *Handler { return &Handler{Service: s} }
+
+// ListConsumersHandler returns consumers, optionally filtered by filePath
+func (h *Handler) ListConsumersHandler(c echo.Context) error {
+	fp := c.QueryParam("filePath")
+	out, err := h.Service.ListConsumers(&models.ConsumerFilter{FilePath: fp})
+	if err != nil {
+		return c.JSON(http.StatusInternalServerError, map[string]string{"error": "Failed to list consumers"})
+	}
+	return c.JSON(http.StatusOK, out)
+}

handlers/scan/scan.go

@@ -6,6 +6,7 @@ import (
 	"vm-server/jobs"
 	"vm-server/models"
 	services "vm-server/services/scan"
+	"vm-server/watcher"
 
 	"github.com/google/uuid"
 	"github.com/labstack/echo/v4"
@@ -15,11 +16,12 @@ import (
 type Handler struct {
 	Service *services.Service
 	Scanner *jobs.Scanner
+	Watcher *watcher.Manager
 }
 
 // NewHandler creates a new scan handler.
-func NewHandler(s *services.Service, sc *jobs.Scanner) *Handler {
-	return &Handler{Service: s, Scanner: sc}
+func NewHandler(s *services.Service, sc *jobs.Scanner, w *watcher.Manager) *Handler {
+	return &Handler{Service: s, Scanner: sc, Watcher: w}
 }
 
 // ScanHandler handles scan related requests.
@@ -41,6 +43,13 @@ func (h *Handler) ScanHandler(c echo.Context) error {
 		return c.JSON(http.StatusInternalServerError, map[string]string{"error": "Failed to create scan job"})
 	}
 
+	// Activate watcher for requested paths (best-effort)
+	if h.Watcher != nil {
+		for _, p := range req.Paths {
+			_ = h.Watcher.ActivatePath(p)
+		}
+	}
+
 	// Run the scan in the background
 	// TODO - Make this a channel so we can perform operations on failed scans as well.
 	go h.Scanner.PerformScan(newJob, &req)

jobs/scan.go

@@ -86,7 +86,7 @@ func (s *Scanner) PerformScan(job *models.ScanJob, req *models.ScanRequest) {
 	// Collect file paths
 	go func() {
 		for _, path := range req.Paths {
-			filepath.WalkDir(path, func(path string, d fs.DirEntry, err error) error {
+			if walkErr := filepath.WalkDir(path, func(path string, d fs.DirEntry, err error) error {
 				if err != nil {
 					log.Printf("Failed to walk directory %s: %v", path, err)
 					return nil // Continue walking
@@ -95,7 +95,9 @@ func (s *Scanner) PerformScan(job *models.ScanJob, req *models.ScanRequest) {
 					filesToScan <- path
 				}
 				return nil
-			})
+			}); walkErr != nil {
+				log.Printf("WalkDir returned error for root %s: %v", path, walkErr)
+			}
 		}
 		close(filesToScan)
 	}()

main.go

@@ -6,12 +6,16 @@ import (
 	"flag"
 	"log"
 	"os"
+	"os/signal"
+	"syscall"
+	consumersHandler "vm-server/handlers/consumers"
 	scanHandler "vm-server/handlers/scan"
 	"vm-server/handlers/secrets"
 	"vm-server/jobs"
 	scanService "vm-server/services/scan"
 	secretsvc "vm-server/services/secrets"
 	memory "vm-server/store/memory"
+	"vm-server/watcher"
 
 	"github.com/labstack/echo/v4"
 	"github.com/labstack/echo/v4/middleware"
@@ -23,6 +27,9 @@ func main() {
 	certFile := flag.String("cert-file", "", "Path to the server certificate file")
 	keyFile := flag.String("key-file", "", "Path to the server key file")
 	port := flag.String("port", "1323", "Port for the server to listen on")
+	watchEnable := flag.Bool("watch-enable", true, "Enable fanotify watcher (Linux only)")
+	watchMount := flag.Bool("watch-mount", false, "Use mount-wide marks for watcher")
+	watchVerbose := flag.Bool("watch-verbose", false, "Verbose watcher logging")
 	flag.Parse()
 
 	// Initialize data store
@@ -45,13 +52,27 @@ func main() {
 	// Initialize job runner
 	scanner := jobs.NewScanner(scanSvc)
 
+	// Initialize fanotify watcher (best-effort)
+	var watchMgr *watcher.Manager
+	if *watchEnable {
+		var werr error
+		watchMgr, werr = watcher.NewManager(scanSvc, watcher.Options{Mount: *watchMount, Verbose: *watchVerbose})
+		if werr != nil {
+			log.Printf("Watcher disabled: %v", werr)
+			watchMgr = nil
+		}
+	} else {
+		log.Printf("Watcher disabled by flag")
+	}
+
 	// Initialize handlers
-	scanHdlr := scanHandler.NewHandler(scanSvc, scanner)
+	scanHdlr := scanHandler.NewHandler(scanSvc, scanner, watchMgr)
 	err = scanner.Cleanup()
 	if err != nil {
 		log.Fatalf("Failed to cleanup: %v", err)
 	}
 	secretHdlr := secrets.NewHandler(secretsSvc, scanSvc)
+	consHdlr := consumersHandler.NewHandler(scanSvc)
 
 	// Echo instance
 	e := echo.New()
@@ -70,6 +91,9 @@ func main() {
 	// Secrets routes
 	apiV1.POST("/secrets/:id/version", secretHdlr.CreateSecretVersionHandler)
 
+	// Consumers routes
+	apiV1.GET("/consumers", consHdlr.ListConsumersHandler)
+
 	// Configure mTLS
 	if *caFile != "" && *certFile != "" && *keyFile != "" {
 		caCert, err := os.ReadFile(*caFile)
@@ -94,4 +118,15 @@ func main() {
 		// Start server without TLS
 		e.Logger.Fatal(e.Start(addr))
 	}
+
+	// Signal handler to stop watcher
+	if watchMgr != nil {
+		sigCh := make(chan os.Signal, 1)
+		signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
+		go func() {
+			<-sigCh
+			log.Printf("Stopping watcher...")
+			watchMgr.Stop()
+		}()
+	}
 }

models/consumer.go

@@ -0,0 +1,29 @@
+package models
+
+import (
+	"time"
+
+	"gorm.io/gorm"
+)
+
+// Consumer represents a process that has accessed a file.
+// Uniqueness is defined by the tuple (FilePath, Comm, Exe, RUID, EUID).
+type Consumer struct {
+	gorm.Model
+	FilePath string `json:"filePath" gorm:"index;uniqueIndex:uniq_consumer,priority:1"`
+	Comm     string `json:"comm" gorm:"uniqueIndex:uniq_consumer,priority:2"`
+	Exe      string `json:"exe" gorm:"uniqueIndex:uniq_consumer,priority:3"`
+	RUID     int    `json:"ruid" gorm:"uniqueIndex:uniq_consumer,priority:4"`
+	EUID     int    `json:"euid" gorm:"uniqueIndex:uniq_consumer,priority:5"`
+}
+
+// FirstSeen returns CreatedAt for convenience.
+func (c *Consumer) FirstSeen() time.Time { return c.CreatedAt }
+
+// LastSeen returns UpdatedAt for convenience.
+func (c *Consumer) LastSeen() time.Time { return c.UpdatedAt }
+
+// ConsumerFilter filters consumers when listing
+type ConsumerFilter struct {
+	FilePath string
+}

services/scan/scan.go

@@ -60,3 +60,12 @@ func (s *Service) DeleteScanEntry(scanEntry *models.ScanEntry) error {
 func (s *Service) ListScanEntries() ([]models.ScanEntry, error) {
 	return s.Store.ListScanEntries()
 }
+
+// Consumers
+func (s *Service) UpsertConsumer(c *models.Consumer) error {
+	return s.Store.UpsertConsumer(c)
+}
+
+func (s *Service) ListConsumers(filter *models.ConsumerFilter) ([]models.Consumer, error) {
+	return s.Store.ListConsumers(filter)
+}

store/gorm/store.go

@@ -3,6 +3,7 @@ package store
 import (
 	"errors"
 	"log"
+	"time"
 	"vm-server/models"
 	"vm-server/store/schema"
 
@@ -26,12 +27,18 @@ func NewStore() (schema.Store, error) {
 
 	log.Println("Database connection established.")
 
-	// Auto-migrate the schema for ScanJob and ScanEntry models.
-	err = db.AutoMigrate(&models.ScanJob{}, &models.ScanEntry{})
+	// Auto-migrate the schema for ScanJob, ScanEntry and Consumer models.
+	err = db.AutoMigrate(&models.ScanJob{}, &models.ScanEntry{}, &models.Consumer{})
 	if err != nil {
 		// Attempt to close the database connection if migration fails.
-		sqlDB, _ := db.DB()
-		sqlDB.Close()
+		sqlDB, dbErr := db.DB()
+		if dbErr != nil {
+			// If we can't get the underlying DB, return the migration error.
+			return nil, err
+		}
+		if closeErr := sqlDB.Close(); closeErr != nil {
+			log.Printf("error closing DB after migration failure: %v", closeErr)
+		}
 		return nil, err
 	}
 
@@ -113,3 +120,34 @@ func (s *StoreGorm) ListScanEntries() ([]models.ScanEntry, error) {
 	result := s.DB.Find(&scanEntries)
 	return scanEntries, result.Error
 }
+
+// UpsertConsumer creates or updates a consumer based on unique tuple.
+func (s *StoreGorm) UpsertConsumer(consumer *models.Consumer) error {
+	var existing models.Consumer
+	tx := s.DB.Where("file_path = ? AND comm = ? AND exe = ? AND ruid = ? AND euid = ?",
+		consumer.FilePath, consumer.Comm, consumer.Exe, consumer.RUID, consumer.EUID).
+		First(&existing)
+	if tx.Error != nil {
+		if errors.Is(tx.Error, gorm.ErrRecordNotFound) {
+			return s.DB.Create(consumer).Error
+		}
+		return tx.Error
+	}
+	existing.UpdatedAt = time.Now()
+	// Keep latest exe/comm if they changed (rare)
+	existing.Comm = consumer.Comm
+	existing.Exe = consumer.Exe
+	return s.DB.Save(&existing).Error
+}
+
+func (s *StoreGorm) ListConsumers(filter *models.ConsumerFilter) ([]models.Consumer, error) {
+	var out []models.Consumer
+	q := s.DB.Model(&models.Consumer{})
+	if filter != nil && filter.FilePath != "" {
+		q = q.Where("file_path = ?", filter.FilePath)
+	}
+	if err := q.Find(&out).Error; err != nil {
+		return nil, err
+	}
+	return out, nil
+}