From 09b6ed6c9ed6eeb08f428548f653249982a2c169 Mon Sep 17 00:00:00 2001 From: Javier Marcos <1271349+javuto@users.noreply.github.com> Date: Mon, 5 Oct 2026 11:30:01 +0200 Subject: [PATCH] Add hot-path indexes and attribute osquery log batches to the authenticated node --- cmd/api/main.go | 7 ++ cmd/tls/handlers/post.go | 2 +- cmd/tls/main.go | 7 ++ pkg/alerts/ingest.go | 8 +- pkg/alerts/matcher.go | 36 ++++-- pkg/alerts/matcher_bench_test.go | 6 +- pkg/alerts/matcher_test.go | 37 +++--- pkg/alerts/node_scope_test.go | 55 +++++++-- pkg/alerts/worker_test.go | 8 +- pkg/auditlog/audit.go | 10 ++ pkg/carves/db.go | 10 ++ pkg/dbutil/indexes.go | 159 ++++++++++++++++++++++++++ pkg/dbutil/indexes_test.go | 188 +++++++++++++++++++++++++++++++ pkg/logging/alerts_hook_test.go | 19 +++- pkg/logging/attribution_test.go | 57 ++++++++++ pkg/logging/bench_alerts_test.go | 10 +- pkg/logging/db.go | 26 ++++- pkg/logging/dispatch.go | 9 +- pkg/logging/logging.go | 8 +- pkg/logging/process.go | 23 ++-- pkg/nodes/indexes_test.go | 36 ++++++ pkg/nodes/metadata_test.go | 37 +++--- pkg/nodes/models.go | 19 ++++ pkg/nodes/nodes.go | 30 +++-- pkg/queries/queries.go | 12 ++ pkg/tags/tags.go | 11 ++ 26 files changed, 718 insertions(+), 112 deletions(-) create mode 100644 pkg/dbutil/indexes.go create mode 100644 pkg/dbutil/indexes_test.go create mode 100644 pkg/logging/attribution_test.go create mode 100644 pkg/nodes/indexes_test.go diff --git a/cmd/api/main.go b/cmd/api/main.go index 0ab265994..92b4dca24 100644 --- a/cmd/api/main.go +++ b/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) diff --git a/cmd/tls/handlers/post.go b/cmd/tls/handlers/post.go index 8a1c8cf48..8bdedf62a 100644 --- a/cmd/tls/handlers/post.go +++ b/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. diff --git a/cmd/tls/main.go b/cmd/tls/main.go index d151af2c6..4cc82132d 100644 --- a/cmd/tls/main.go +++ b/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. diff --git a/pkg/alerts/ingest.go b/pkg/alerts/ingest.go index dd7c0c458..8b5ceba3f 100644 --- a/pkg/alerts/ingest.go +++ b/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. diff --git a/pkg/alerts/matcher.go b/pkg/alerts/matcher.go index 24c1c23ea..2ad9ee598 100644 --- a/pkg/alerts/matcher.go +++ b/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 diff --git a/pkg/alerts/matcher_bench_test.go b/pkg/alerts/matcher_bench_test.go index 447c6f3e4..1c9ad22a7 100644 --- a/pkg/alerts/matcher_bench_test.go +++ b/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) } } diff --git a/pkg/alerts/matcher_test.go b/pkg/alerts/matcher_test.go index efc040df6..5f24f10b4 100644 --- a/pkg/alerts/matcher_test.go +++ b/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 { diff --git a/pkg/alerts/node_scope_test.go b/pkg/alerts/node_scope_test.go index 0f32f7c47..34b320236 100644 --- a/pkg/alerts/node_scope_test.go +++ b/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) } diff --git a/pkg/alerts/worker_test.go b/pkg/alerts/worker_test.go index 2ccf2882c..7d42b8fc4 100644 --- a/pkg/alerts/worker_test.go +++ b/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, "") } diff --git a/pkg/auditlog/audit.go b/pkg/auditlog/audit.go index 9de463039..c6898f333 100644 --- a/pkg/auditlog/audit.go +++ b/pkg/auditlog/audit.go @@ -4,6 +4,7 @@ import ( "fmt" "time" + "github.com/jmpsec/osctrl/pkg/dbutil" "github.com/rs/zerolog/log" "gorm.io/gorm" ) @@ -549,3 +550,12 @@ func (m *AuditLogManager) GetBySeverityEnv(severity, envID uint) ([]AuditLog, er } return logs, nil } + +// Indexes are the secondary indexes for audit_logs beyond those in the struct +// tags, created by dbutil.EnsureIndexes: the audit page lists entries newest +// first. +func Indexes() []dbutil.Index { + return []dbutil.Index{ + {Model: &AuditLog{}, Name: "idx_audit_logs_created_at", Columns: []string{"created_at"}}, + } +} diff --git a/pkg/carves/db.go b/pkg/carves/db.go index d97861f87..da3fb9428 100644 --- a/pkg/carves/db.go +++ b/pkg/carves/db.go @@ -3,6 +3,7 @@ package carves import ( "time" + "github.com/jmpsec/osctrl/pkg/dbutil" "gorm.io/gorm" ) @@ -41,3 +42,12 @@ type CarvedBlock struct { Carver string EnvironmentID uint } + +// Indexes are the secondary indexes for carved_files beyond those in the +// struct tags, created by dbutil.EnsureIndexes: the carves list looks up each +// carve's files by query name within an environment. +func Indexes() []dbutil.Index { + return []dbutil.Index{ + {Model: &CarvedFile{}, Name: "idx_carved_files_env_query", Columns: []string{"environment_id", "query_name"}}, + } +} diff --git a/pkg/dbutil/indexes.go b/pkg/dbutil/indexes.go new file mode 100644 index 000000000..6f6109a7f --- /dev/null +++ b/pkg/dbutil/indexes.go @@ -0,0 +1,159 @@ +package dbutil + +import ( + "errors" + "fmt" + "strings" + + "github.com/rs/zerolog/log" + "gorm.io/gorm" +) + +// mysqlTextPrefix is the prefix length MySQL indexes TEXT columns with: it can +// only index those by prefix. 191 utf8mb4 characters is the length GORM itself +// gives indexed strings, and stays within InnoDB's key size in composites. +const mysqlTextPrefix = 191 + +// Index is a secondary index created outside GORM's struct tags. +// +// Tags would be the usual way, but they are unsafe for string columns that +// were created without an index: GORM's MySQL driver types an untagged string +// as longtext and an indexed one as varchar(191), so tagging an existing +// column makes AutoMigrate issue an ALTER that rebuilds the table and fails, +// or truncates, on any value longer than 191 characters. Tags also cannot +// reach the timestamps of an embedded gorm.Model. +type Index struct { + // Model is the GORM model of the table, which also supplies any table + // prefix from the naming strategy. + Model any + // Name must be unique in the schema. + Name string + // Columns are database column names, in index order. + Columns []string +} + +// EnsureIndexes creates whichever of indexes do not exist yet, and is a no-op +// for the rest. It is safe to run from several processes at once: osctrl-api +// and osctrl-tls both run it at startup. +// +// On Postgres the build is CONCURRENTLY, so writers are never blocked however +// large the table is. A concurrent build that is interrupted — the process +// restarting mid-build, say — leaves an invalid index that Postgres keeps +// maintaining on every write but never reads; those are dropped and rebuilt, +// unless another session is still building them. +// +// MySQL builds secondary indexes online. SQLite holds its single write lock +// for the build, which is short on the databases it is used for. +func EnsureIndexes(db *gorm.DB, indexes ...Index) error { + var errs []error + for _, idx := range indexes { + if err := ensureIndex(db, idx); err != nil { + errs = append(errs, fmt.Errorf("index %s: %w", idx.Name, err)) + } + } + return errors.Join(errs...) +} + +// BuildIndexes runs EnsureIndexes without holding up the caller: in the +// background on Postgres and MySQL, where a build can take minutes on a large +// table but does not block writers, and inline on SQLite, whose databases are +// small and which allows one writer at a time regardless. A failure is logged +// rather than returned — the indexes speed reads up; nothing depends on them +// existing — and the next startup tries again. +func BuildIndexes(db *gorm.DB, indexes ...Index) { + run := func() { + if err := EnsureIndexes(db, indexes...); err != nil { + log.Warn().Err(err).Msg("creating database indexes") + } + } + if db.Name() == "sqlite" { + run() + return + } + go run() +} + +func ensureIndex(db *gorm.DB, idx Index) error { + if !db.Migrator().HasIndex(idx.Model, idx.Name) { + return createIndex(db, idx) + } + if db.Name() != "postgres" { + return nil + } + var valid, building bool + err := db.Raw(`SELECT i.indisvalid, + EXISTS (SELECT 1 FROM pg_stat_progress_create_index p WHERE p.index_relid = c.oid) + FROM pg_class c + JOIN pg_index i ON i.indexrelid = c.oid + JOIN pg_namespace n ON n.oid = c.relnamespace + WHERE c.relname = ? AND n.nspname = current_schema()`, idx.Name).Row().Scan(&valid, &building) + if err != nil { + return fmt.Errorf("check validity: %w", err) + } + if valid || building { + return nil + } + log.Warn().Str("index", idx.Name).Msg("rebuilding invalid index left by an interrupted concurrent build") + if err := db.Exec("DROP INDEX CONCURRENTLY IF EXISTS " + quote(db, idx.Name)).Error; err != nil { + return fmt.Errorf("drop invalid index: %w", err) + } + return createIndex(db, idx) +} + +func createIndex(db *gorm.DB, idx Index) error { + stmt := &gorm.Statement{DB: db} + if err := stmt.Parse(idx.Model); err != nil { + return fmt.Errorf("parse model: %w", err) + } + textColumns, err := mysqlTextColumns(db, idx.Model) + if err != nil { + return err + } + columns := make([]string, len(idx.Columns)) + for i, c := range idx.Columns { + columns[i] = quote(db, c) + if textColumns[c] { + columns[i] += fmt.Sprintf("(%d)", mysqlTextPrefix) + } + } + on := fmt.Sprintf("%s ON %s (%s)", quote(db, idx.Name), quote(db, stmt.Schema.Table), strings.Join(columns, ", ")) + var sql string + switch db.Name() { + case "postgres": + sql = "CREATE INDEX CONCURRENTLY IF NOT EXISTS " + on + case "mysql": + // MySQL has no IF NOT EXISTS here; ensureIndex already checked. + sql = "CREATE INDEX " + on + default: + sql = "CREATE INDEX IF NOT EXISTS " + on + } + return db.Exec(sql).Error +} + +// mysqlTextColumns reports which columns of model are TEXT or BLOB on MySQL, +// the only ones that need a prefix length. A prefix on a short varchar is an +// error, so this reads the live column types instead of assuming them. Other +// dialects get an empty set. +func mysqlTextColumns(db *gorm.DB, model any) (map[string]bool, error) { + if db.Name() != "mysql" { + return nil, nil + } + types, err := db.Migrator().ColumnTypes(model) + if err != nil { + return nil, fmt.Errorf("read column types: %w", err) + } + text := make(map[string]bool) + for _, t := range types { + name := strings.ToLower(t.DatabaseTypeName()) + if strings.Contains(name, "text") || strings.Contains(name, "blob") { + text[t.Name()] = true + } + } + return text, nil +} + +func quote(db *gorm.DB, name string) string { + var b strings.Builder + db.QuoteTo(&b, name) + return b.String() +} diff --git a/pkg/dbutil/indexes_test.go b/pkg/dbutil/indexes_test.go new file mode 100644 index 000000000..33d30c9d4 --- /dev/null +++ b/pkg/dbutil/indexes_test.go @@ -0,0 +1,188 @@ +package dbutil_test + +import ( + "fmt" + "os" + "slices" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/require" + "gorm.io/driver/mysql" + "gorm.io/driver/postgres" + "gorm.io/driver/sqlite" + "gorm.io/gorm" + "gorm.io/gorm/schema" + + "github.com/jmpsec/osctrl/pkg/auditlog" + "github.com/jmpsec/osctrl/pkg/carves" + "github.com/jmpsec/osctrl/pkg/dbutil" + "github.com/jmpsec/osctrl/pkg/logging" + "github.com/jmpsec/osctrl/pkg/nodes" + "github.com/jmpsec/osctrl/pkg/queries" + "github.com/jmpsec/osctrl/pkg/tags" +) + +type widget struct { + ID uint + Name string + Kind string + Code string `gorm:"type:varchar(36)"` + CreatedAt time.Time +} + +func openSQLite(t *testing.T, cfg *gorm.Config) *gorm.DB { + t.Helper() + db, err := gorm.Open(sqlite.Open("file:"+strings.ReplaceAll(t.Name(), "/", "_")+"?mode=memory&cache=shared"), cfg) + require.NoError(t, err) + return db +} + +func countExecs(t *testing.T, db *gorm.DB) *int { + t.Helper() + n := new(int) + require.NoError(t, db.Callback().Raw().After("gorm:raw").Register("count_index_ddl_"+t.Name(), func(*gorm.DB) { *n++ })) + return n +} + +func TestEnsureIndexesCreatesMissingOnce(t *testing.T) { + db := openSQLite(t, &gorm.Config{}) + require.NoError(t, db.AutoMigrate(&widget{})) + idx := dbutil.Index{Model: &widget{}, Name: "idx_widgets_kind_created", Columns: []string{"kind", "created_at"}} + + require.NoError(t, dbutil.EnsureIndexes(db, idx)) + require.True(t, db.Migrator().HasIndex(&widget{}, idx.Name)) + + execs := countExecs(t, db) + require.NoError(t, dbutil.EnsureIndexes(db, idx)) + require.Zero(t, *execs, "an existing index must not be rebuilt on every startup") +} + +func TestEnsureIndexesHonoursTablePrefix(t *testing.T) { + db := openSQLite(t, &gorm.Config{NamingStrategy: schema.NamingStrategy{TablePrefix: "pfx_"}}) + require.NoError(t, db.AutoMigrate(&widget{})) + idx := dbutil.Index{Model: &widget{}, Name: "idx_pfx_widgets_name", Columns: []string{"name"}} + + require.NoError(t, dbutil.EnsureIndexes(db, idx)) + require.True(t, db.Migrator().HasIndex(&widget{}, idx.Name)) + var table string + require.NoError(t, db.Raw(`SELECT tbl_name FROM sqlite_master WHERE type = 'index' AND name = ?`, idx.Name).Scan(&table).Error) + require.Equal(t, "pfx_widgets", table) +} + +// One bad index must not stop the others from being created. +func TestEnsureIndexesContinuesPastFailures(t *testing.T) { + db := openSQLite(t, &gorm.Config{}) + require.NoError(t, db.AutoMigrate(&widget{})) + + err := dbutil.EnsureIndexes(db, + dbutil.Index{Model: &widget{}, Name: "idx_widgets_missing", Columns: []string{"no_such_column"}}, + dbutil.Index{Model: &widget{}, Name: "idx_widgets_name", Columns: []string{"name"}}, + ) + require.Error(t, err) + require.Contains(t, err.Error(), "idx_widgets_missing") + require.True(t, db.Migrator().HasIndex(&widget{}, "idx_widgets_name")) +} + +func osctrlIndexes() []dbutil.Index { + return slices.Concat(nodes.Indexes(), queries.Indexes(), auditlog.Indexes(), + carves.Indexes(), tags.Indexes(), logging.Indexes()) +} + +func osctrlModels() []any { + return []any{ + &nodes.OsqueryNode{}, &queries.DistributedQuery{}, &queries.NodeQuery{}, &auditlog.AuditLog{}, + &carves.CarvedFile{}, &tags.TaggedNode{}, + &logging.OsqueryStatusData{}, &logging.OsqueryResultData{}, &logging.OsqueryQueryData{}, + } +} + +// Every index osctrl declares must build against the real models — this is +// where a misspelt column name would surface. +func TestOsctrlIndexesBuild(t *testing.T) { + db := openSQLite(t, &gorm.Config{}) + require.NoError(t, db.AutoMigrate(osctrlModels()...)) + + require.NoError(t, dbutil.EnsureIndexes(db, osctrlIndexes()...)) + names := map[string]bool{} + for _, idx := range osctrlIndexes() { + require.Falsef(t, names[idx.Name], "index name %s declared twice", idx.Name) + names[idx.Name] = true + require.Truef(t, db.Migrator().HasIndex(idx.Model, idx.Name), "index %s missing", idx.Name) + } +} + +// Optional real-engine checks, each on its own prefixed tables. +func externalDB(t *testing.T, backend string) (*gorm.DB, string) { + t.Helper() + dsn := os.Getenv("OSCTRL_TEST_" + backend + "_DSN") + if dsn == "" { + t.Skip("set OSCTRL_TEST_" + backend + "_DSN to test this engine") + } + dialect := postgres.Open(dsn) + if backend == "MYSQL" { + dialect = mysql.Open(dsn) + } + prefix := fmt.Sprintf("idx_%d_", time.Now().UnixNano()) + db, err := gorm.Open(dialect, &gorm.Config{NamingStrategy: schema.NamingStrategy{TablePrefix: prefix}}) + require.NoError(t, err) + return db, prefix +} + +func TestOsctrlIndexesBuildExternalBackend(t *testing.T) { + for _, backend := range []string{"POSTGRES", "MYSQL"} { + t.Run(backend, func(t *testing.T) { + db, prefix := externalDB(t, backend) + models := osctrlModels() + t.Cleanup(func() { require.NoError(t, db.Migrator().DropTable(models...)) }) + require.NoError(t, db.AutoMigrate(models...)) + // Index names are schema-wide on Postgres, so the shared names + // would collide with any other copy of these tables there. + indexes := osctrlIndexes() + for i := range indexes { + indexes[i].Name = prefix + indexes[i].Name + } + + require.NoError(t, dbutil.EnsureIndexes(db, indexes...)) + for _, idx := range indexes { + require.Truef(t, db.Migrator().HasIndex(idx.Model, idx.Name), "index %s missing", idx.Name) + } + require.NoError(t, dbutil.EnsureIndexes(db, indexes...), "a second run must be a no-op") + }) + } +} + +// MySQL can only index TEXT by prefix, and rejects a prefix on a short +// varchar: one index over both kinds proves the prefix goes on the TEXT column +// alone. +func TestEnsureIndexesMySQLTextPrefix(t *testing.T) { + db, _ := externalDB(t, "MYSQL") + t.Cleanup(func() { require.NoError(t, db.Migrator().DropTable(&widget{})) }) + require.NoError(t, db.AutoMigrate(&widget{})) + idx := dbutil.Index{Model: &widget{}, Name: "idx_widgets_name_code", Columns: []string{"name", "code"}} + + require.NoError(t, dbutil.EnsureIndexes(db, idx)) + require.True(t, db.Migrator().HasIndex(&widget{}, idx.Name)) +} + +// A concurrent build that fails leaves an invalid index Postgres keeps +// maintaining but never reads. A failed CREATE UNIQUE INDEX CONCURRENTLY over +// duplicate rows is the documented way to produce one. +func TestEnsureIndexesRebuildsInvalidPostgresIndex(t *testing.T) { + db, prefix := externalDB(t, "POSTGRES") + t.Cleanup(func() { require.NoError(t, db.Migrator().DropTable(&widget{})) }) + require.NoError(t, db.AutoMigrate(&widget{})) + require.NoError(t, db.Create(&[]widget{{Name: "dup"}, {Name: "dup"}}).Error) + + name := prefix + "idx_widgets_name" + table := prefix + "widgets" + require.Error(t, db.Exec(fmt.Sprintf(`CREATE UNIQUE INDEX CONCURRENTLY %q ON %q (name)`, name, table)).Error) + var valid bool + require.NoError(t, db.Raw(`SELECT i.indisvalid FROM pg_index i JOIN pg_class c ON c.oid = i.indexrelid WHERE c.relname = ?`, name).Scan(&valid).Error) + require.False(t, valid, "fixture: the failed build should have left an invalid index") + + require.NoError(t, dbutil.EnsureIndexes(db, dbutil.Index{Model: &widget{}, Name: name, Columns: []string{"name"}})) + require.NoError(t, db.Raw(`SELECT i.indisvalid FROM pg_index i JOIN pg_class c ON c.oid = i.indexrelid WHERE c.relname = ?`, name).Scan(&valid).Error) + require.True(t, valid, "the invalid index must have been rebuilt") +} diff --git a/pkg/logging/alerts_hook_test.go b/pkg/logging/alerts_hook_test.go index 8e8ff6f36..99a25d038 100644 --- a/pkg/logging/alerts_hook_test.go +++ b/pkg/logging/alerts_hook_test.go @@ -11,6 +11,9 @@ import ( "gorm.io/gorm" ) +// hookNode is the node the hook tests' batches are authenticated as. +var hookNode = nodes.OsqueryNode{ID: 1, UUID: "NODE-A"} + // recordingMatcher captures hook invocations for the ingest tests. type recordingMatcher struct { resultBatches []int @@ -18,21 +21,24 @@ type recordingMatcher struct { queryResults []string lastEnvID uint lastEnvName string + lastNodeUUID string lastResultLogs []types.LogResultData lastStatusLogs []types.LogStatusData } -func (m *recordingMatcher) MatchResultLogs(envID uint, environment string, logs []types.LogResultData) { +func (m *recordingMatcher) MatchResultLogs(envID uint, environment, nodeUUID string, logs []types.LogResultData) { m.resultBatches = append(m.resultBatches, len(logs)) m.lastEnvID = envID m.lastEnvName = environment + m.lastNodeUUID = nodeUUID m.lastResultLogs = logs } -func (m *recordingMatcher) MatchStatusLogs(envID uint, environment string, logs []types.LogStatusData) { +func (m *recordingMatcher) MatchStatusLogs(envID uint, environment, nodeUUID string, logs []types.LogStatusData) { m.statusBatches = append(m.statusBatches, len(logs)) m.lastEnvID = envID m.lastEnvName = environment + m.lastNodeUUID = nodeUUID m.lastStatusLogs = logs } @@ -48,7 +54,7 @@ func TestProcessLogsAlertHookResult(t *testing.T) { logger := testLoggerWithMatcher(t, &recordingMatcher{}) data := json.RawMessage(`[{"name":"file_events","columns":{"path":"/etc/shadow"},"hostIdentifier":"NODE-A"}]`) - returned := logger.ProcessLogs(data, types.ResultLog, 3, "prod", "10.0.0.1", len(data), false) + returned := logger.ProcessLogs(hookNode, data, types.ResultLog, 3, "prod", "10.0.0.1", len(data), false) if len(returned) != 1 { t.Fatalf("ProcessLogs must still return decoded results: %d", len(returned)) @@ -63,6 +69,9 @@ func TestProcessLogsAlertHookResult(t *testing.T) { if m.lastResultLogs[0].HostIdentifier != "NODE-A" { t.Fatalf("unexpected entry passed to matcher: %+v", m.lastResultLogs[0]) } + if m.lastNodeUUID != hookNode.UUID { + t.Fatalf("matcher must be told the authenticated sender, got %q", m.lastNodeUUID) + } } // TestProcessLogsAlertHookStatus verifies the status-log tap decodes the @@ -71,7 +80,7 @@ func TestProcessLogsAlertHookStatus(t *testing.T) { logger := testLoggerWithMatcher(t, &recordingMatcher{}) data := json.RawMessage(`[{"severity":"2","message":"error: bad thing happened","hostIdentifier":"NODE-A"}]`) - logger.ProcessLogs(data, types.StatusLog, 3, "prod", "10.0.0.1", len(data), false) + logger.ProcessLogs(hookNode, data, types.StatusLog, 3, "prod", "10.0.0.1", len(data), false) m := logger.Alerts.(*recordingMatcher) if len(m.statusBatches) != 1 { @@ -91,7 +100,7 @@ func TestProcessLogsNoMatcherIsNoop(t *testing.T) { logger := testLoggerWithMatcher(t, nil) data := json.RawMessage(`[{"name":"x","columns":{"a":"b"},"hostIdentifier":"U"}]`) - returned := logger.ProcessLogs(data, types.ResultLog, 1, "dev", "10.0.0.1", len(data), false) + returned := logger.ProcessLogs(hookNode, data, types.ResultLog, 1, "dev", "10.0.0.1", len(data), false) if len(returned) != 1 { t.Fatalf("no-matcher path must not change behavior: %d", len(returned)) } diff --git a/pkg/logging/attribution_test.go b/pkg/logging/attribution_test.go new file mode 100644 index 000000000..e2ecf008e --- /dev/null +++ b/pkg/logging/attribution_test.go @@ -0,0 +1,57 @@ +package logging + +import ( + "encoding/json" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/jmpsec/osctrl/pkg/nodes" + "github.com/jmpsec/osctrl/pkg/queries" + "github.com/jmpsec/osctrl/pkg/types" +) + +// A log batch is authenticated by node_key as one node, but every entry in it +// carries a hostIdentifier the sender writes. Nothing the batch produces — the +// metadata update, the stored rows, alert hits — may follow that claim, or any +// enrolled node could overwrite another's metadata, plant log lines on its +// page, and frame it in alerts. +func TestProcessLogsAttributesToAuthenticatedNode(t *testing.T) { + logDB := newTestLoggerDB(t) + db := logDB.Database.Conn + attacker := nodes.OsqueryNode{UUID: "ATTACKER", EnvironmentID: 1, Environment: "prod", Hostname: "attacker-host"} + victim := nodes.OsqueryNode{UUID: "VICTIM", EnvironmentID: 1, Environment: "prod", Hostname: "victim-host"} + nodeMgr := nodes.CreateNodes(db) + require.NoError(t, db.Create(&attacker).Error) + require.NoError(t, db.Create(&victim).Error) + + matcher := &recordingMatcher{} + logger := &LoggerTLS{Logging: "db", Nodes: nodeMgr, Queries: queries.CreateQueries(db), Alerts: matcher} + logger.ReplaceExporters(map[uint]*MultiExporter{0: NewMultiExporter(logDB)}) + + status := json.RawMessage(`[{"hostIdentifier":"VICTIM","severity":"2","message":"planted","decorations":{"hostname":"spoofed"}}]`) + logger.ProcessLogs(attacker, status, types.StatusLog, 1, "prod", "10.0.0.9", len(status), false) + require.Equal(t, "ATTACKER", matcher.lastNodeUUID, "status alerts must be attributed to the sender") + + result := json.RawMessage(`[{"name":"pack_x","hostIdentifier":"VICTIM","columns":{"a":"b"},"decorations":{"hostname":"spoofed"}}]`) + logger.ProcessLogs(attacker, result, types.ResultLog, 1, "prod", "10.0.0.9", len(result), false) + require.Equal(t, "ATTACKER", matcher.lastNodeUUID, "result alerts must be attributed to the sender") + + // The victim's row is untouched; the batch describes its sender. + var gotVictim, gotAttacker nodes.OsqueryNode + require.NoError(t, db.First(&gotVictim, victim.ID).Error) + require.NoError(t, db.First(&gotAttacker, attacker.ID).Error) + require.Equal(t, "victim-host", gotVictim.Hostname, "a claimed hostIdentifier overwrote another node's metadata") + require.Zero(t, gotVictim.BytesReceived) + require.Equal(t, "spoofed", gotAttacker.Hostname) + require.Equal(t, len(status)+len(result), gotAttacker.BytesReceived) + + // Stored rows are filed under the sender, never the claimed node. + for _, model := range []any{&OsqueryStatusData{}, &OsqueryResultData{}} { + var claimed, sender int64 + require.NoError(t, db.Model(model).Where("uuid = ?", "VICTIM").Count(&claimed).Error) + require.NoError(t, db.Model(model).Where("uuid = ?", "ATTACKER").Count(&sender).Error) + require.Zerof(t, claimed, "%T rows filed under the claimed node", model) + require.EqualValuesf(t, 1, sender, "%T rows not filed under the sender", model) + } +} diff --git a/pkg/logging/bench_alerts_test.go b/pkg/logging/bench_alerts_test.go index 4fd56e908..7602b49d5 100644 --- a/pkg/logging/bench_alerts_test.go +++ b/pkg/logging/bench_alerts_test.go @@ -95,11 +95,11 @@ type noopMatcher struct { calls int } -func (m *noopMatcher) MatchResultLogs(envID uint, environment string, logs []types.LogResultData) { +func (m *noopMatcher) MatchResultLogs(envID uint, environment, nodeUUID string, logs []types.LogResultData) { m.calls += len(logs) } -func (m *noopMatcher) MatchStatusLogs(envID uint, environment string, logs []types.LogStatusData) { +func (m *noopMatcher) MatchStatusLogs(envID uint, environment, nodeUUID string, logs []types.LogStatusData) { m.calls += len(logs) } @@ -123,7 +123,7 @@ func benchmarkProcessLogs(b *testing.B, matcher AlertMatcher) { b.ReportAllocs() b.ResetTimer() for i := 0; i < b.N; i++ { - _ = logger.ProcessLogs(data, "result", 1, "prod", "10.0.0.1", len(data), false) + _ = logger.ProcessLogs(hookNode, data, "result", 1, "prod", "10.0.0.1", len(data), false) } } @@ -156,7 +156,7 @@ func BenchmarkProcessLogsStatusBaseline(b *testing.B) { b.ReportAllocs() b.ResetTimer() for i := 0; i < b.N; i++ { - _ = logger.ProcessLogs(data, "status", 1, "prod", "10.0.0.1", len(data), false) + _ = logger.ProcessLogs(hookNode, data, "status", 1, "prod", "10.0.0.1", len(data), false) } } @@ -177,6 +177,6 @@ func BenchmarkProcessLogsStatusMatcher(b *testing.B) { b.ReportAllocs() b.ResetTimer() for i := 0; i < b.N; i++ { - _ = logger.ProcessLogs(data, "status", 1, "prod", "10.0.0.1", len(data), false) + _ = logger.ProcessLogs(hookNode, data, "status", 1, "prod", "10.0.0.1", len(data), false) } } diff --git a/pkg/logging/db.go b/pkg/logging/db.go index b5e9d3d88..6505f48ff 100644 --- a/pkg/logging/db.go +++ b/pkg/logging/db.go @@ -86,6 +86,9 @@ func CreateLoggerDB(backend *backend.DBManager) (*LoggerDB, error) { if err := backend.Conn.AutoMigrate(&OsqueryQueryData{}); err != nil { log.Fatal().Msgf("Failed to AutoMigrate table (osquery_query_data): %v", err) } + // Here rather than with the other tables' indexes: the log tables may live + // in a separate logger database, which a log-sink reload can also open. + dbutil.BuildIndexes(backend.Conn, Indexes()...) return l, nil } @@ -141,7 +144,9 @@ func (logDB *LoggerDB) Log(logType string, data []byte, environment, uuid string } } -// Status - Function that sends JSON status logs to the configured DB +// Status - Function that sends JSON status logs to the configured DB. Rows are +// filed under uuid, the node the batch was authenticated as, not under each +// entry's sender-controlled hostIdentifier. func (logDB *LoggerDB) Status(data []byte, environment, uuid string, debug bool) { // Parse JSON var logs []types.LogStatusData @@ -153,7 +158,7 @@ func (logDB *LoggerDB) Status(data []byte, environment, uuid string, debug bool) for _, l := range logs { entries = append(entries, OsqueryStatusData{ Model: stamp(), - UUID: strings.ToUpper(l.HostIdentifier), + UUID: strings.ToUpper(uuid), Environment: environment, Line: strconv.Itoa(int(l.Line)), Message: l.Message, @@ -165,7 +170,8 @@ func (logDB *LoggerDB) Status(data []byte, environment, uuid string, debug bool) insertLogRows(logDB.Database.Conn, entries, "status") } -// Result - Function that sends JSON result logs to the configured DB +// Result - Function that sends JSON result logs to the configured DB, filed +// under uuid as for Status. func (logDB *LoggerDB) Result(data []byte, environment, uuid string, debug bool) { // Parse JSON logs, err := parseResultLogs(data) @@ -177,7 +183,7 @@ func (logDB *LoggerDB) Result(data []byte, environment, uuid string, debug bool) for _, l := range logs { entries = append(entries, OsqueryResultData{ Model: stamp(), - UUID: strings.ToUpper(l.HostIdentifier), + UUID: strings.ToUpper(uuid), Environment: environment, Name: l.Name, Action: l.Action, @@ -594,3 +600,15 @@ func (logDB *LoggerDB) CleanQueryLogs(entries int64) error { } return nil } + +// Indexes are the secondary indexes for the log tables beyond those in the +// struct tags, created by dbutil.EnsureIndexes. A node's status and result logs +// are read newest first, and a query's results oldest first; without +// (uuid, created_at) a node's log page reads every row it has ever sent. +func Indexes() []dbutil.Index { + return []dbutil.Index{ + {Model: &OsqueryStatusData{}, Name: "idx_osquery_status_data_uuid_created", Columns: []string{"uuid", "created_at"}}, + {Model: &OsqueryResultData{}, Name: "idx_osquery_result_data_uuid_created", Columns: []string{"uuid", "created_at"}}, + {Model: &OsqueryQueryData{}, Name: "idx_osquery_query_data_name_created", Columns: []string{"name", "created_at"}}, + } +} diff --git a/pkg/logging/dispatch.go b/pkg/logging/dispatch.go index 52c218f92..c07af12e1 100644 --- a/pkg/logging/dispatch.go +++ b/pkg/logging/dispatch.go @@ -13,15 +13,18 @@ import ( // LoggerTLS.LogWithEnv). The string environment is forwarded as // metadata for exporters that embed it (Splunk source type, DB row, // etc.). -func (l *LoggerTLS) DispatchLogs(data []byte, uuid, logType string, envID uint, environment string, metadata nodes.NodeMetadata, debug bool) { +// +// node is the authenticated sender: its metadata is updated and its UUID is +// what the exporters file the batch under. +func (l *LoggerTLS) DispatchLogs(data []byte, node nodes.OsqueryNode, logType string, envID uint, environment string, metadata nodes.NodeMetadata, debug bool) { // Use metadata to update record - if err := l.Nodes.UpdateMetadataByUUID(uuid, envID, metadata); err != nil { + if err := l.Nodes.UpdateMetadata(node.ID, metadata); err != nil { log.Err(err).Msg("error updating metadata") } if debug { log.Debug().Msgf("dispatching logs to %s", l.Logging) } - l.LogWithEnv(logType, data, envID, environment, uuid, debug) + l.LogWithEnv(logType, data, envID, environment, node.UUID, debug) } // DispatchQueries - Helper to dispatch queries. Uses the node's diff --git a/pkg/logging/logging.go b/pkg/logging/logging.go index c677d5e28..12eb2eeb4 100644 --- a/pkg/logging/logging.go +++ b/pkg/logging/logging.go @@ -17,10 +17,12 @@ import ( // feature-off state (nil interface + nil receiver = zero cost). type AlertMatcher interface { // MatchResultLogs is called with the decoded result-log batch - // after parse, before dispatch. It must not block. - MatchResultLogs(envID uint, environment string, logs []types.LogResultData) + // after parse, before dispatch. It must not block. nodeUUID is the + // node the batch was authenticated as; the entries' hostIdentifier is + // sender-controlled and must not be used to attribute hits. + MatchResultLogs(envID uint, environment, nodeUUID string, logs []types.LogResultData) // MatchStatusLogs is the status-log counterpart. - MatchStatusLogs(envID uint, environment string, logs []types.LogStatusData) + MatchStatusLogs(envID uint, environment, nodeUUID string, logs []types.LogStatusData) // MatchQueryResult is called per query in ProcessLogQueryResult. MatchQueryResult(envID uint, environment, queryName string, result json.RawMessage, status int, message string) } diff --git a/pkg/logging/process.go b/pkg/logging/process.go index 8093dec21..0f276bac3 100644 --- a/pkg/logging/process.go +++ b/pkg/logging/process.go @@ -24,7 +24,14 @@ func parseResultLogs(data json.RawMessage) ([]types.LogResultData, error) { // ProcessLogs processes and dispatches logs. Result entries are returned so // callers can reuse the decoded batch for secondary consumers such as posture. // envID selects the environment-scoped exporter set. -func (l *LoggerTLS) ProcessLogs(data json.RawMessage, logType string, envID uint, environment, ipaddress string, dataLen int, debug bool) []types.LogResultData { +// +// node is the node the batch was authenticated as, by node_key. Everything the +// batch produces — the metadata update, the exported rows, alert hits — is +// attributed to it. The entries' hostIdentifier is never used for that: the +// sender writes it, so trusting it let any enrolled node overwrite another +// node's metadata, plant log rows under another node, or fire and dodge +// node-scoped alerts. +func (l *LoggerTLS) ProcessLogs(node nodes.OsqueryNode, data json.RawMessage, logType string, envID uint, environment, ipaddress string, dataLen int, debug bool) []types.LogResultData { // Parse log to extract metadata var logs []types.LogGenericData var resultLogs []types.LogResultData @@ -73,15 +80,15 @@ func (l *LoggerTLS) ProcessLogs(data json.RawMessage, logType string, envID uint if l.Alerts != nil { switch logType { case types.ResultLog: - l.Alerts.MatchResultLogs(envID, environment, resultLogs) + l.Alerts.MatchResultLogs(envID, environment, node.UUID, resultLogs) case types.StatusLog: - l.Alerts.MatchStatusLogs(envID, environment, statusLogs) + l.Alerts.MatchStatusLogs(envID, environment, node.UUID, statusLogs) } } - // Iterate through received messages to extract metadata - var uuid, hostname, localname, username, osqueryuser, confighash, daemonhash, osqueryversion string + // Iterate through received messages to extract the metadata the node + // decorates its logs with. + var hostname, localname, username, osqueryuser, confighash, daemonhash, osqueryversion string for _, l := range logs { - uuid = metadataVerification(uuid, l.HostIdentifier) hostname = metadataVerification(hostname, l.Decorations.Hostname) localname = metadataVerification(localname, l.Decorations.LocalHostname) username = metadataVerification(username, l.Decorations.Username) @@ -91,7 +98,7 @@ func (l *LoggerTLS) ProcessLogs(data json.RawMessage, logType string, envID uint osqueryversion = metadataVerification(osqueryversion, l.Decorations.OsqueryVersion) } if debug { - log.Debug().Msgf("metadata and dispatch for %s", uuid) + log.Debug().Msgf("metadata and dispatch for %s", node.UUID) } metadata := nodes.NodeMetadata{ IPAddress: ipaddress, @@ -105,7 +112,7 @@ func (l *LoggerTLS) ProcessLogs(data json.RawMessage, logType string, envID uint BytesReceived: dataLen, } // Dispatch logs and update metadata - l.DispatchLogs(data, uuid, logType, envID, environment, metadata, debug) + l.DispatchLogs(data, node, logType, envID, environment, metadata, debug) return resultLogs } diff --git a/pkg/nodes/indexes_test.go b/pkg/nodes/indexes_test.go new file mode 100644 index 000000000..32c1ba8f3 --- /dev/null +++ b/pkg/nodes/indexes_test.go @@ -0,0 +1,36 @@ +package nodes + +import ( + "slices" + "sync" + "testing" + + "gorm.io/gorm/schema" +) + +// Check-ins and log batches rewrite these columns constantly. While no indexed +// column changes, Postgres applies those updates in place (HOT) without +// touching any index; indexing one of them would make every check-in maintain +// every index on the table. +func TestIndexesLeaveCheckinColumnsUnindexed(t *testing.T) { + hot := []string{"last_seen", "ip_address", "bytes_received", "last_query_read"} + for _, idx := range Indexes() { + for _, c := range idx.Columns { + if slices.Contains(hot, c) { + t.Errorf("index %s covers %s, which every check-in rewrites", idx.Name, c) + } + } + } + // The struct tags must not index them either. + s, err := schema.Parse(&OsqueryNode{}, &sync.Map{}, schema.NamingStrategy{}) + if err != nil { + t.Fatal(err) + } + for _, idx := range s.ParseIndexes() { + for _, f := range idx.Fields { + if slices.Contains(hot, f.DBName) { + t.Errorf("struct tag index %s covers %s, which every check-in rewrites", idx.Name, f.DBName) + } + } + } +} diff --git a/pkg/nodes/metadata_test.go b/pkg/nodes/metadata_test.go index 2d1ec234b..2fce36b25 100644 --- a/pkg/nodes/metadata_test.go +++ b/pkg/nodes/metadata_test.go @@ -16,14 +16,14 @@ import ( "gorm.io/gorm/schema" ) -func TestUpdateMetadataByUUID(t *testing.T) { +func TestUpdateMetadata(t *testing.T) { db, err := gorm.Open(sqlite.Open("file:"+t.Name()+"?mode=memory&cache=shared"), &gorm.Config{}) require.NoError(t, err) - testUpdateMetadataByUUID(t, db) + testUpdateMetadata(t, db) } // Optional real-engine checks; each uses and removes its own prefixed table. -func TestUpdateMetadataByUUIDExternalBackend(t *testing.T) { +func TestUpdateMetadataExternalBackend(t *testing.T) { for _, backend := range []string{"POSTGRES", "MYSQL"} { t.Run(backend, func(t *testing.T) { dsn := os.Getenv("OSCTRL_TEST_" + backend + "_DSN") @@ -39,17 +39,18 @@ func TestUpdateMetadataByUUIDExternalBackend(t *testing.T) { }}) require.NoError(t, err) t.Cleanup(func() { require.NoError(t, db.Migrator().DropTable(&OsqueryNode{})) }) - testUpdateMetadataByUUID(t, db) + testUpdateMetadata(t, db) }) } } -func testUpdateMetadataByUUID(t *testing.T, db *gorm.DB) { +func testUpdateMetadata(t *testing.T, db *gorm.DB) { t.Helper() require.NoError(t, db.AutoMigrate(&OsqueryNode{})) repo := NodeManager{DB: db} - // The same UUID enrolled in two environments has a live row in each. + // The same UUID enrolled in two environments has a live row in each; an + // update by primary key must touch only the one it names. rows := []OsqueryNode{ {UUID: "NODE-A", EnvironmentID: 1, Hostname: "old-host", Username: "keep-me", BytesReceived: 100}, {UUID: "NODE-A", EnvironmentID: 2, Hostname: "other-env", BytesReceived: 7}, @@ -61,9 +62,8 @@ func testUpdateMetadataByUUID(t *testing.T, db *gorm.DB) { require.NoError(t, db.Callback().Query().After("gorm:query").Register("count_metadata_selects", func(*gorm.DB) { selects++ })) require.NoError(t, db.Callback().Update().After("gorm:update").Register("count_metadata_updates", func(*gorm.DB) { updates++ })) - // Lower-case UUID, as host identifiers can arrive; empty fields must not - // blank out what is stored. - require.NoError(t, repo.UpdateMetadataByUUID("node-a", 1, NodeMetadata{ + // Empty fields must not blank out what is stored. + require.NoError(t, repo.UpdateMetadata(rows[0].ID, NodeMetadata{ Hostname: "new-host", ConfigHash: "abc123", OsqueryVersion: "5.23.1", @@ -85,24 +85,24 @@ func testUpdateMetadataByUUID(t *testing.T, db *gorm.DB) { require.Equal(t, "keep-me", target.Username, "an empty field must leave the stored value alone") require.Equal(t, 125, target.BytesReceived) - // Only the row of the authenticated environment is touched. + // Only the row named by ID is touched. require.Equal(t, "other-env", got["NODE-A/2"].Hostname, "same UUID in another environment must be untouched") require.Equal(t, 7, got["NODE-A/2"].BytesReceived) require.Equal(t, "unrelated", got["NODE-B/1"].Hostname) require.Equal(t, 50, got["NODE-B/1"].BytesReceived) - // Unknown nodes still report not-found, as the read-first version did. - err := repo.UpdateMetadataByUUID("NODE-MISSING", 1, NodeMetadata{BytesReceived: 1}) - require.Error(t, err) + // Unknown and deleted nodes report not-found, as the read-first version did. + err := repo.UpdateMetadata(rows[len(rows)-1].ID+100, NodeMetadata{BytesReceived: 1}) require.True(t, errors.Is(err, gorm.ErrRecordNotFound), "got %v", err) - err = repo.UpdateMetadataByUUID("NODE-A", 99, NodeMetadata{BytesReceived: 1}) - require.True(t, errors.Is(err, gorm.ErrRecordNotFound), "wrong environment must not match: got %v", err) + require.NoError(t, db.Delete(&rows[2]).Error) + err = repo.UpdateMetadata(rows[2].ID, NodeMetadata{BytesReceived: 1}) + require.True(t, errors.Is(err, gorm.ErrRecordNotFound), "a deleted node must not be updated: got %v", err) } // Log batches from one node are processed in parallel goroutines. Reading // bytes_received and writing back a sum lost increments whenever two // overlapped; the SQL-side increment must account for every byte. -func TestUpdateMetadataByUUIDConcurrentBytes(t *testing.T) { +func TestUpdateMetadataConcurrentBytes(t *testing.T) { dir := t.TempDir() // A file database so connections share state and SQLite serialises // writers for real, rather than one in-memory connection doing it all. @@ -110,14 +110,15 @@ func TestUpdateMetadataByUUIDConcurrentBytes(t *testing.T) { require.NoError(t, err) require.NoError(t, db.AutoMigrate(&OsqueryNode{})) repo := NodeManager{DB: db} - require.NoError(t, db.Create(&OsqueryNode{UUID: "NODE-A", EnvironmentID: 1}).Error) + node := OsqueryNode{UUID: "NODE-A", EnvironmentID: 1} + require.NoError(t, db.Create(&node).Error) const writers, each = 8, 25 var wg sync.WaitGroup for range writers { wg.Go(func() { for range each { - if err := repo.UpdateMetadataByUUID("NODE-A", 1, NodeMetadata{BytesReceived: 3}); err != nil { + if err := repo.UpdateMetadata(node.ID, NodeMetadata{BytesReceived: 3}); err != nil { t.Errorf("update: %v", err) return } diff --git a/pkg/nodes/models.go b/pkg/nodes/models.go index dc1ca5083..a48b038fc 100644 --- a/pkg/nodes/models.go +++ b/pkg/nodes/models.go @@ -4,6 +4,8 @@ import ( "time" "gorm.io/gorm" + + "github.com/jmpsec/osctrl/pkg/dbutil" ) // OsqueryNode as abstraction of a node @@ -86,3 +88,20 @@ type NodeMetadata struct { PlatformVersion string BytesReceived int } + +// Indexes are the secondary indexes for osquery_nodes beyond those in the +// struct tags, created by dbutil.EnsureIndexes. environment serves the +// per-environment counts and paged lists; hostname and localname let the +// uuid-or-hostname-or-localname lookups use an index for every branch. +// +// last_seen, ip_address and bytes_received are deliberately not indexed: +// every check-in or log batch rewrites them, and while no indexed column +// changes, Postgres can apply those updates in place (HOT) without touching +// any index on the table. +func Indexes() []dbutil.Index { + return []dbutil.Index{ + {Model: &OsqueryNode{}, Name: "idx_osquery_nodes_environment", Columns: []string{"environment"}}, + {Model: &OsqueryNode{}, Name: "idx_osquery_nodes_hostname", Columns: []string{"hostname"}}, + {Model: &OsqueryNode{}, Name: "idx_osquery_nodes_localname", Columns: []string{"localname"}}, + } +} diff --git a/pkg/nodes/nodes.go b/pkg/nodes/nodes.go index 35181c2c7..245fb1162 100644 --- a/pkg/nodes/nodes.go +++ b/pkg/nodes/nodes.go @@ -384,20 +384,18 @@ func (n *NodeManager) GetOsqueryVersionCounts() ([]OsqueryVersionCount, error) { return GetOsqueryVersionCounts(n.DB) } -// UpdateMetadataByUUID records the metadata osquery decorates its log batches -// with on the node of that UUID enrolled in envID. +// UpdateMetadata records the metadata osquery decorates its log batches with +// on node nodeID: the node the batch was authenticated as, by node_key. The +// caller must not derive nodeID from the batch itself — its hostIdentifier is +// whatever the sender chose to write. // -// It runs on every log batch every node sends, so it is a single UPDATE. -// bytes_received is incremented in SQL instead of being read and written back, -// which also stops two log batches from the same node, processed concurrently, -// from losing each other's bytes. The other columns are assigned whenever the -// batch carries a value: storing an unchanged value leaves the row as it was, -// so there is no need to read the row first to compare. -// -// Scoping by environment matters because a UUID enrolled in two environments -// has a live row in each; the caller has already authenticated the batch -// against envID, so that is the row it describes. -func (n *NodeManager) UpdateMetadataByUUID(uuid string, envID uint, metadata NodeMetadata) error { +// It runs on every log batch every node sends, so it is a single UPDATE by +// primary key. bytes_received is incremented in SQL instead of being read and +// written back, which also stops two log batches from the same node, processed +// concurrently, from losing each other's bytes. The other columns are assigned +// whenever the batch carries a value: storing an unchanged value leaves the row +// as it was, so there is no need to read the row first to compare. +func (n *NodeManager) UpdateMetadata(nodeID uint, metadata NodeMetadata) error { updates := map[string]interface{}{ "bytes_received": gorm.Expr("bytes_received + ?", metadata.BytesReceived), } @@ -415,14 +413,12 @@ func (n *NodeManager) UpdateMetadataByUUID(uuid string, envID uint, metadata Nod updates[column] = value } } - result := n.DB.Model(&OsqueryNode{}). - Where("uuid = ? AND environment_id = ?", strings.ToUpper(uuid), envID). - Updates(updates) + result := n.DB.Model(&OsqueryNode{}).Where("id = ?", nodeID).Updates(updates) if result.Error != nil { return fmt.Errorf("update metadata %w", result.Error) } if result.RowsAffected == 0 { - return fmt.Errorf("update metadata for node %s: %w", uuid, gorm.ErrRecordNotFound) + return fmt.Errorf("update metadata for node %d: %w", nodeID, gorm.ErrRecordNotFound) } return nil } diff --git a/pkg/queries/queries.go b/pkg/queries/queries.go index 4c28a8c79..8c6ef6de0 100644 --- a/pkg/queries/queries.go +++ b/pkg/queries/queries.go @@ -813,3 +813,15 @@ func queryResourceIDs(extraData, resourceKey string) (uint, uint, bool) { resourceID := fields[resourceKey] return sessionID, resourceID, sessionID != 0 && resourceID != 0 } + +// Indexes are the secondary indexes for the query tables beyond those in the +// struct tags, created by dbutil.EnsureIndexes. (query_id, status) answers +// "is any target of this query still pending" on every result osctrl-tls +// records; (environment_id, type, created_at) serves the per-environment query +// lists and active counts, newest first. +func Indexes() []dbutil.Index { + return []dbutil.Index{ + {Model: &NodeQuery{}, Name: "idx_node_queries_query_status", Columns: []string{"query_id", "status"}}, + {Model: &DistributedQuery{}, Name: "idx_distributed_queries_env_type_created", Columns: []string{"environment_id", "type", "created_at"}}, + } +} diff --git a/pkg/tags/tags.go b/pkg/tags/tags.go index d6cbf62ae..447e3f15a 100644 --- a/pkg/tags/tags.go +++ b/pkg/tags/tags.go @@ -6,6 +6,7 @@ import ( "strings" "time" + "github.com/jmpsec/osctrl/pkg/dbutil" "github.com/jmpsec/osctrl/pkg/nodes" "github.com/rs/zerolog/log" "gorm.io/gorm" @@ -633,3 +634,13 @@ func (m *TagManager) CountTaggedNodes(tags []AdminTag) (TagCounter, error) { } return tagCounter, nil } + +// Indexes are the secondary indexes for tagged_nodes, created by +// dbutil.EnsureIndexes: every node list loads the tags of its page by node, +// and the tags page counts nodes per tag. +func Indexes() []dbutil.Index { + return []dbutil.Index{ + {Model: &TaggedNode{}, Name: "idx_tagged_nodes_node_id", Columns: []string{"node_id"}}, + {Model: &TaggedNode{}, Name: "idx_tagged_nodes_admin_tag_id", Columns: []string{"admin_tag_id"}}, + } +}