Add hot-path indexes and attribute osquery log batches to the authenticated node

cmd/api/main.go

@@ -9,6 +9,7 @@ import (
 	"os"
 	"os/signal"
 	"path/filepath"
+	"slices"
 	"strconv"
 	"strings"
 	"syscall"
@@ -24,6 +25,7 @@ import (
 	"github.com/jmpsec/osctrl/pkg/carves"
 	"github.com/jmpsec/osctrl/pkg/config"
 	"github.com/jmpsec/osctrl/pkg/console"
+	"github.com/jmpsec/osctrl/pkg/dbutil"
 	"github.com/jmpsec/osctrl/pkg/environments"
 	"github.com/jmpsec/osctrl/pkg/events"
 	"github.com/jmpsec/osctrl/pkg/fileexplorer"
@@ -551,6 +553,11 @@ func osctrlAPIService() {
 	if err != nil {
 		log.Fatal().Msgf("Error initializing audit log manager - %v", err)
 	}
+	// Secondary indexes for the hot read paths, beyond the struct tags. Built
+	// in the background, so a large table does not hold up startup; osctrl-api
+	// and osctrl-tls both run this and are safe to race.
+	dbutil.BuildIndexes(db.Conn, slices.Concat(
+		nodes.Indexes(), queries.Indexes(), auditlog.Indexes(), carves.Indexes(), tags.Indexes())...)
 	// Load osquery tables schema (best-effort; an empty slice is fine if the file doesn't exist)
 	if flagParams.Osquery.TablesFile != "" {
 		log.Info().Msgf("Loading osquery tables from %s", flagParams.Osquery.TablesFile)

cmd/tls/handlers/post.go

@@ -354,7 +354,7 @@ func (h *HandlersTLS) LogHandler(w http.ResponseWriter, r *http.Request) {
 		// Process logs and update metadata
 		go func() {
 			start := time.Now()
-			results := h.Logs.ProcessLogs(t.Data, t.LogType, env.ID, env.Name, utils.GetIP(r), len(body), env.DebugHTTP)
+			results := h.Logs.ProcessLogs(node, t.Data, t.LogType, env.ID, env.Name, utils.GetIP(r), len(body), env.DebugHTTP)
 			duration := time.Since(start).Seconds()
 			logProcessDuration.WithLabelValues(string(env.UUID), t.LogType).Observe(duration)
 			// Count ERROR-severity status logs into the activity rollup.

cmd/tls/main.go

@@ -8,6 +8,7 @@ import (
 	"net/http"
 	"os"
 	"path/filepath"
+	"slices"
 	"strconv"
 	"strings"
 	"time"
@@ -20,6 +21,7 @@ import (
 	"github.com/jmpsec/osctrl/pkg/cache"
 	"github.com/jmpsec/osctrl/pkg/carves"
 	"github.com/jmpsec/osctrl/pkg/config"
+	"github.com/jmpsec/osctrl/pkg/dbutil"
 	"github.com/jmpsec/osctrl/pkg/environments"
 	"github.com/jmpsec/osctrl/pkg/events"
 	"github.com/jmpsec/osctrl/pkg/health"
@@ -456,6 +458,11 @@ func osctrlService() {
 	if err != nil {
 		log.Fatal().Msgf("error initializing audit log manager: %v", err)
 	}
+	// Secondary indexes for the hot read paths, beyond the struct tags. Built
+	// in the background, so a large table does not hold up startup; osctrl-api
+	// and osctrl-tls both run this and are safe to race.
+	dbutil.BuildIndexes(db.Conn, slices.Concat(
+		nodes.Indexes(), queries.Indexes(), auditlog.Indexes(), carves.Indexes(), tags.Indexes())...)
 	restartCh := make(chan struct{}, 1)
 	// Per-IP rate limit on /enroll. Defaults to bursts of 20 per minute,
 	// idle eviction after 10 minutes.

pkg/alerts/ingest.go

@@ -29,19 +29,19 @@ func NewIngestMatcher(store *Store, worker *Worker) *IngestMatcher {
 }
 
 // MatchResultLogs implements logging.AlertMatcher.
-func (m *IngestMatcher) MatchResultLogs(envID uint, environment string, logs []types.LogResultData) {
+func (m *IngestMatcher) MatchResultLogs(envID uint, environment, nodeUUID string, logs []types.LogResultData) {
 	if m == nil {
 		return
 	}
-	m.worker.Enqueue(m.store.Snapshot().MatchResultLogs(envID, environment, logs))
+	m.worker.Enqueue(m.store.Snapshot().MatchResultLogs(envID, environment, nodeUUID, logs))
 }
 
 // MatchStatusLogs implements logging.AlertMatcher.
-func (m *IngestMatcher) MatchStatusLogs(envID uint, environment string, logs []types.LogStatusData) {
+func (m *IngestMatcher) MatchStatusLogs(envID uint, environment, nodeUUID string, logs []types.LogStatusData) {
 	if m == nil {
 		return
 	}
-	m.worker.Enqueue(m.store.Snapshot().MatchStatusLogs(envID, environment, logs))
+	m.worker.Enqueue(m.store.Snapshot().MatchStatusLogs(envID, environment, nodeUUID, logs))
 }
 
 // MatchQueryResult implements logging.AlertMatcher.

pkg/alerts/matcher.go

@@ -45,14 +45,20 @@ func (r *compiledRule) ruleApplies(envID uint) bool {
 // MatchResultLogs evaluates result-log entries against rules scoped to
 // envID. The columns of each entry (or its snapshot rows) are matched
 // field-scoped or across all fields depending on the rule.
-func (rs *RuleSet) MatchResultLogs(envID uint, environment string, logs []types.LogResultData) []Hit {
+//
+// nodeUUID is the node osctrl-tls authenticated the batch as, by node_key.
+// Hits are attributed to it, and node-scoped rules are checked against it,
+// never against the entries' hostIdentifier: the sender writes that field, so
+// trusting it would let a node frame another one, or evade a rule scoped to
+// itself by claiming to be someone else.
+func (rs *RuleSet) MatchResultLogs(envID uint, environment, nodeUUID string, logs []types.LogResultData) []Hit {
 	if len(rs.result) == 0 || len(logs) == 0 {
 		return nil
 	}
 	var hits []Hit
 	for i := range logs {
 		entry := &logs[i]
-		uuid := entry.HostIdentifier
+		uuid := nodeUUID
 		fields := resultFields(entry)
 		lowered := make(map[string]string, len(fields))
 		for j := range rs.result {
@@ -84,15 +90,16 @@ func (rs *RuleSet) MatchResultLogs(envID uint, environment string, logs []types.
 
 // MatchStatusLogs evaluates status-log entries against rules scoped to
 // envID. Entries below the rule's severity floor are skipped before any
-// pattern work happens.
-func (rs *RuleSet) MatchStatusLogs(envID uint, environment string, logs []types.LogStatusData) []Hit {
+// pattern work happens. nodeUUID is the authenticated sender, as for
+// MatchResultLogs.
+func (rs *RuleSet) MatchStatusLogs(envID uint, environment, nodeUUID string, logs []types.LogStatusData) []Hit {
 	if len(rs.status) == 0 || len(logs) == 0 {
 		return nil
 	}
 	var hits []Hit
 	for i := range logs {
 		entry := &logs[i]
-		uuid := entry.HostIdentifier
+		uuid := nodeUUID
 		lowered := make(map[string]string, 3)
 		for j := range rs.status {
 			r := &rs.status[j]
@@ -265,11 +272,20 @@ func matchStatus(r *compiledRule, entry *types.LogStatusData, lowered map[string
 			return false, ""
 		}
 	}
-	return matchFields(r, map[string]string{
-		"message":  entry.Message,
-		"filename": entry.Filename,
-		"version":  entry.Version,
-	}, lowered)
+	// Fields in a fixed order rather than through matchFields' map, whose
+	// iteration order is random: when several fields match — always the case
+	// for the empty pattern of the node page's "any error" preset — the
+	// reported detail must be the message, not whichever field came first.
+	for _, f := range [...]struct{ key, value string }{
+		{"message", entry.Message},
+		{"filename", entry.Filename},
+		{"version", entry.Version},
+	} {
+		if matched, detail := matchSingle(r, f.value, f.key, lowered); matched {
+			return true, detail
+		}
+	}
+	return false, ""
 }
 
 // substringDetail is the context a substring match reports. An empty

pkg/alerts/matcher_bench_test.go

@@ -72,7 +72,7 @@ func BenchmarkMatchResultLogs20x500(b *testing.B) {
 	b.ReportAllocs()
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
-		_ = rs.MatchResultLogs(NoEnvironmentID, "dev", logs)
+		_ = rs.MatchResultLogs(NoEnvironmentID, "dev", testNode, logs)
 	}
 }
 
@@ -82,7 +82,7 @@ func BenchmarkMatchResultLogs50x1000(b *testing.B) {
 	b.ReportAllocs()
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
-		_ = rs.MatchResultLogs(NoEnvironmentID, "dev", logs)
+		_ = rs.MatchResultLogs(NoEnvironmentID, "dev", testNode, logs)
 	}
 }
 
@@ -92,6 +92,6 @@ func BenchmarkMatchResultLogsEmptyRules(b *testing.B) {
 	b.ReportAllocs()
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
-		_ = rs.MatchResultLogs(NoEnvironmentID, "dev", logs)
+		_ = rs.MatchResultLogs(NoEnvironmentID, "dev", testNode, logs)
 	}
 }

pkg/alerts/matcher_test.go

@@ -35,6 +35,11 @@ func compileForTest(t *testing.T, rules ...AlertRule) *RuleSet {
 	return rs
 }
 
+// testNode is the node the matcher tests' batches are authenticated as. The
+// entries' own hostIdentifier values are deliberately different: they are
+// sender-controlled and must not decide attribution.
+const testNode = "NODE-A"
+
 func resultEntry(uuid, name string, columns map[string]string) types.LogResultData {
 	raw, _ := json.Marshal(columns)
 	return types.LogResultData{
@@ -62,19 +67,19 @@ func TestMatchResultLogs(t *testing.T) {
 		resultEntry("UUID2", "file_events", map[string]string{"path": "/etc/shadow"}),
 		resultEntry("UUID3", "file_events", map[string]string{"path": "/etc/hostname"}),
 	}
-	hits := rs.MatchResultLogs(NoEnvironmentID, "dev", logs)
+	hits := rs.MatchResultLogs(NoEnvironmentID, "dev", testNode, logs)
 	if len(hits) != 2 {
 		t.Fatalf("expected 2 hits, got %d: %+v", len(hits), hits)
 	}
-	// Hit 1: case-insensitive any-field substring on UUID1.
-	if hits[0].RuleName != "any-substring" || hits[0].NodeUUID != "UUID1" {
+	// Hit 1: case-insensitive any-field substring on the first entry.
+	if hits[0].RuleName != "any-substring" || hits[0].NodeUUID != testNode {
 		t.Fatalf("unexpected hit 0: %+v", hits[0])
 	}
-	// Hit 2: scoped regex path match on UUID2.
-	if hits[1].RuleName != "scoped-regex" || hits[1].NodeUUID != "UUID2" {
+	// Hit 2: scoped regex path match on the second entry.
+	if hits[1].RuleName != "scoped-regex" || hits[1].NodeUUID != testNode {
 		t.Fatalf("unexpected hit 1: %+v", hits[1])
 	}
-	if hits[1].Entity != "UUID2:file_events" {
+	if hits[1].Entity != testNode+":file_events" {
 		t.Fatalf("entity should include query name: %q", hits[1].Entity)
 	}
 }
@@ -86,7 +91,7 @@ func TestMatchResultLogsScopedFieldMissing(t *testing.T) {
 	// Entry has no "path" column — scoped rule must not fall back to
 	// other fields.
 	logs := []types.LogResultData{resultEntry("U", "processes", map[string]string{"name": "/etc value in wrong field"})}
-	if hits := rs.MatchResultLogs(NoEnvironmentID, "dev", logs); len(hits) != 0 {
+	if hits := rs.MatchResultLogs(NoEnvironmentID, "dev", testNode, logs); len(hits) != 0 {
 		t.Fatalf("scoped rule matched absent field: %+v", hits)
 	}
 }
@@ -98,7 +103,7 @@ func TestMatchResultLogsSnapshot(t *testing.T) {
 	snap := []map[string]string{{"username": "root", "host": "box"}}
 	raw, _ := json.Marshal(snap)
 	logs := []types.LogResultData{{HostIdentifier: "U", Name: "logged_in_users", Snapshot: raw}}
-	hits := rs.MatchResultLogs(NoEnvironmentID, "dev", logs)
+	hits := rs.MatchResultLogs(NoEnvironmentID, "dev", testNode, logs)
 	if len(hits) != 1 {
 		t.Fatalf("expected snapshot match, got %+v", hits)
 	}
@@ -116,7 +121,7 @@ func TestMatchStatusLogs(t *testing.T) {
 		statusEntry("U3", 1, "warning: scheduler backoff"),        // matches warn-or-worse + info-ok
 		statusEntry("U4", 2, "error: different message entirely"), // matches any-error + info? no line... any-error yes
 	}
-	hits := rs.MatchStatusLogs(NoEnvironmentID, "dev", logs)
+	hits := rs.MatchStatusLogs(NoEnvironmentID, "dev", testNode, logs)
 	// U1: info-ok (1). U2: any-error + warn-or-worse + info-ok? "line" not
 	// in message... info-ok pattern is "line" and message has no "line":
 	// so U2 matches any-error + warn-or-worse (2). U3: warn-or-worse (1).
@@ -135,12 +140,12 @@ func TestMatchResultLogsEnvScoping(t *testing.T) {
 	logs := []types.LogResultData{resultEntry("U", "t", map[string]string{"f": "match-me"})}
 
 	// env 7: both rules apply.
-	hits := rs.MatchResultLogs(7, "prod", logs)
+	hits := rs.MatchResultLogs(7, "prod", testNode, logs)
 	if len(hits) != 2 {
 		t.Fatalf("env 7 expected 2 hits, got %d", len(hits))
 	}
 	// other envs: only the global rule.
-	hits = rs.MatchResultLogs(9, "other", logs)
+	hits = rs.MatchResultLogs(9, "other", testNode, logs)
 	if len(hits) != 1 || hits[0].RuleName != "global" {
 		t.Fatalf("env 9 expected global-only hit, got %+v", hits)
 	}
@@ -158,8 +163,10 @@ func TestMatchStatusLogsSeverityFloor(t *testing.T) {
 		statusEntry("U2", 1, "match at warn"), // below floor — skipped
 		statusEntry("U3", 2, "match at error"),
 	}
-	hits := rs.MatchStatusLogs(NoEnvironmentID, "dev", logs)
-	if len(hits) != 1 || hits[0].NodeUUID != "U3" {
+	hits := rs.MatchStatusLogs(NoEnvironmentID, "dev", testNode, logs)
+	// All three lines contain the pattern; only the error-level one clears
+	// the floor.
+	if len(hits) != 1 || hits[0].NodeUUID != testNode {
 		t.Fatalf("severity floor not enforced: %+v", hits)
 	}
 }
@@ -192,10 +199,10 @@ func TestMatchQueryResultFailedQuery(t *testing.T) {
 
 func TestMatchNoRulesNoWork(t *testing.T) {
 	rs := NewStore().Snapshot()
-	if hits := rs.MatchResultLogs(NoEnvironmentID, "dev", []types.LogResultData{resultEntry("U", "t", map[string]string{"a": "b"})}); hits != nil {
+	if hits := rs.MatchResultLogs(NoEnvironmentID, "dev", testNode, []types.LogResultData{resultEntry("U", "t", map[string]string{"a": "b"})}); hits != nil {
 		t.Fatalf("empty snapshot must not hit: %+v", hits)
 	}
-	if hits := rs.MatchStatusLogs(NoEnvironmentID, "dev", nil); hits != nil {
+	if hits := rs.MatchStatusLogs(NoEnvironmentID, "dev", testNode, nil); hits != nil {
 		t.Fatalf("empty snapshot must not hit: %+v", hits)
 	}
 	if hits := rs.MatchQueryResult(NoEnvironmentID, "dev", "q", nil, 0, ""); hits != nil {

pkg/alerts/node_scope_test.go

@@ -10,17 +10,17 @@ import (
 )
 
 // TestMatchNodeScopedResultRule pins the node-scoping contract: a rule
-// with NodeUUID set only matches that node's entries.
+// with NodeUUID set only matches batches from that node.
 func TestMatchNodeScopedResultRule(t *testing.T) {
 	rs := compileForTest(t,
 		AlertRule{Model: ruleWithID(1), Name: "scoped", Source: SourceResultLog, MatchType: MatchTypeSubstring, MatchValue: "needle", NodeUUID: "UUID-SCOPED"},
 		AlertRule{Model: ruleWithID(2), Name: "global", Source: SourceResultLog, MatchType: MatchTypeSubstring, MatchValue: "needle"},
 	)
-	logs := []types.LogResultData{
-		resultEntry("UUID-SCOPED", "file_events", map[string]string{"path": "needle-here"}),
-		resultEntry("UUID-OTHER", "file_events", map[string]string{"path": "needle-here"}),
+	batch := func(uuid string) []types.LogResultData {
+		return []types.LogResultData{resultEntry(uuid, "file_events", map[string]string{"path": "needle-here"})}
 	}
-	hits := rs.MatchResultLogs(7, "prod", logs)
+	hits := append(rs.MatchResultLogs(7, "prod", "UUID-SCOPED", batch("UUID-SCOPED")),
+		rs.MatchResultLogs(7, "prod", "UUID-OTHER", batch("UUID-OTHER"))...)
 	if len(hits) != 3 {
 		// scoped fires on UUID-SCOPED only (1), global fires on both (2)
 		t.Fatalf("expected 3 hits (1 scoped + 2 global), got %d: %+v", len(hits), hits)
@@ -44,16 +44,47 @@ func TestMatchNodeScopedStatusRule(t *testing.T) {
 	rs := compileForTest(t,
 		AlertRule{Model: ruleWithID(1), Name: "scoped-status", Source: SourceStatusLog, MatchType: MatchTypeSubstring, MatchValue: "error", NodeUUID: "UUID-SCOPED"},
 	)
-	logs := []types.LogStatusData{
-		statusEntry("UUID-SCOPED", 2, "error: mine"),
-		statusEntry("UUID-OTHER", 2, "error: not mine"),
-	}
-	hits := rs.MatchStatusLogs(7, "prod", logs)
+	hits := append(rs.MatchStatusLogs(7, "prod", "UUID-SCOPED", []types.LogStatusData{statusEntry("UUID-SCOPED", 2, "error: mine")}),
+		rs.MatchStatusLogs(7, "prod", "UUID-OTHER", []types.LogStatusData{statusEntry("UUID-OTHER", 2, "error: not mine")})...)
 	if len(hits) != 1 || hits[0].NodeUUID != "UUID-SCOPED" {
 		t.Fatalf("scoped status rule must only fire for its node: %+v", hits)
 	}
 }
 
+// A batch's entries carry a hostIdentifier the sender writes. A node that
+// claims to be another must neither fire that node's scoped rules (framing
+// it) nor slip past rules scoped to itself (evading them): attribution is the
+// node the batch was authenticated as.
+func TestMatchIgnoresClaimedHostIdentifier(t *testing.T) {
+	rs := compileForTest(t,
+		AlertRule{Model: ruleWithID(1), Name: "victim-result", Source: SourceResultLog, MatchType: MatchTypeSubstring, MatchValue: "needle", NodeUUID: "VICTIM"},
+		AlertRule{Model: ruleWithID(2), Name: "attacker-result", Source: SourceResultLog, MatchType: MatchTypeSubstring, MatchValue: "needle", NodeUUID: "ATTACKER"},
+		AlertRule{Model: ruleWithID(3), Name: "victim-status", Source: SourceStatusLog, MatchType: MatchTypeSubstring, MatchValue: "error", NodeUUID: "VICTIM"},
+		AlertRule{Model: ruleWithID(4), Name: "attacker-status", Source: SourceStatusLog, MatchType: MatchTypeSubstring, MatchValue: "error", NodeUUID: "ATTACKER"},
+	)
+	// Authenticated as ATTACKER, every entry claiming to be VICTIM.
+	results := rs.MatchResultLogs(7, "prod", "ATTACKER", []types.LogResultData{
+		resultEntry("VICTIM", "file_events", map[string]string{"path": "needle"}),
+	})
+	statuses := rs.MatchStatusLogs(7, "prod", "ATTACKER", []types.LogStatusData{
+		statusEntry("VICTIM", 2, "error: spoofed"),
+	})
+	for _, h := range append(results, statuses...) {
+		if h.NodeUUID != "ATTACKER" {
+			t.Fatalf("hit attributed to the claimed node instead of the sender: %+v", h)
+		}
+		if strings.HasPrefix(h.RuleName, "victim") {
+			t.Fatalf("a spoofed hostIdentifier fired another node's rule: %+v", h)
+		}
+	}
+	if len(results) != 1 || results[0].RuleName != "attacker-result" {
+		t.Fatalf("the sender's own result rule must still fire: %+v", results)
+	}
+	if len(statuses) != 1 || statuses[0].RuleName != "attacker-status" {
+		t.Fatalf("the sender's own status rule must still fire: %+v", statuses)
+	}
+}
+
 // TestInactiveWatcherNodeScopedRule: a node_inactive rule scoped to one
 // UUID fires only for that node's transition.
 func TestInactiveWatcherNodeScopedRule(t *testing.T) {
@@ -95,9 +126,9 @@ func TestMatchNodeScopedErrorRule(t *testing.T) {
 		statusEntry("WATCH-U9", 2, "query failed to execute"),
 		statusEntry("WATCH-U9", 1, "a warning nobody asked to be paged for"),
 		statusEntry("WATCH-U9", 0, "an informational line"),
-		statusEntry("OTHER-NODE", 2, "someone else's error"),
 	}
-	hits := rs.MatchStatusLogs(0, "prod", logs)
+	hits := append(rs.MatchStatusLogs(0, "prod", "WATCH-U9", logs),
+		rs.MatchStatusLogs(0, "prod", "OTHER-NODE", []types.LogStatusData{statusEntry("OTHER-NODE", 2, "someone else's error")})...)
 	if len(hits) != 1 {
 		t.Fatalf("expected exactly the node's error line to hit, got %+v", hits)
 	}

pkg/alerts/worker_test.go

@@ -211,16 +211,16 @@ func TestIngestMatcherAdapts(t *testing.T) {
 	store.Publish(rs)
 
 	matcher := NewIngestMatcher(store, worker)
-	matcher.MatchResultLogs(1, "dev", []types.LogResultData{resultEntry("node-1", "file_events", map[string]string{"path": "needle here"})})
-	matcher.MatchStatusLogs(1, "dev", []types.LogStatusData{statusEntry("node-1", 2, "an error occurred")})
+	matcher.MatchResultLogs(1, "dev", "NODE-1", []types.LogResultData{resultEntry("node-1", "file_events", map[string]string{"path": "needle here"})})
+	matcher.MatchStatusLogs(1, "dev", "NODE-1", []types.LogStatusData{statusEntry("node-1", 2, "an error occurred")})
 	if sink.count() != 2 {
 		t.Fatalf("expected 2 dispatches through adapter, got %d", sink.count())
 	}
 
 	// nil matcher must be safe
 	var nilMatcher *IngestMatcher
-	nilMatcher.MatchResultLogs(1, "dev", nil)
-	nilMatcher.MatchStatusLogs(1, "dev", nil)
+	nilMatcher.MatchResultLogs(1, "dev", "NODE-1", nil)
+	nilMatcher.MatchStatusLogs(1, "dev", "NODE-1", nil)
 	nilMatcher.MatchQueryResult(1, "dev", "q", nil, 0, "")
 }