447 lines
17 KiB
JavaScript
447 lines
17 KiB
JavaScript
/**
|
|
* Security Event Store
|
|
*
|
|
* Unified JSONL store for security-relevant events from any source:
|
|
* - api : DashCaddy API audit events (extends src/security/audit-logger.js)
|
|
* - caddy : Caddy reverse-proxy access log events (parsed by tail-worker)
|
|
* - fail2ban : SSH brute-force bans (read from /var/log/fail2ban.log)
|
|
* - shared-bans : IP blocklist promotion events (read from /var/log/shared-bans-promote.log)
|
|
* - syslog : (v2) Generic syslog messages
|
|
* - agent : (v2) Events from a remote DCA (DashCaddy Agent) binary
|
|
*
|
|
* Storage format: JSONL (one JSON object per line). Why JSONL not JSON-array?
|
|
* - Append-only writes are O(1) — no read-modify-write race
|
|
* - Partial reads on crash (last line may be corrupt, but earlier lines survive)
|
|
* - Trivial to grep/jq for forensics
|
|
* - Easy to tail from a remote source
|
|
*
|
|
* Query strategy: in-memory index with file-backed persistence. For <100k events
|
|
* this is fine. Beyond that we'd switch to SQLite — see BACKLOG.md (DC-???) to track.
|
|
*
|
|
* The store also emits events to subscribers (in-process EventEmitter), so the
|
|
* dashboard can do live tailing via Server-Sent Events in v2.
|
|
*/
|
|
|
|
const fs = require('fs');
|
|
const path = require('path');
|
|
const crypto = require('crypto');
|
|
const { EventEmitter } = require('events');
|
|
const platformPaths = require('../../platform-paths');
|
|
|
|
const EVENT_STORE_FILE = process.env.SECURITY_EVENT_LOG_FILE
|
|
|| path.join(platformPaths.dataDir, 'security-events.jsonl');
|
|
|
|
const MAX_EVENTS_IN_MEMORY = parseInt(process.env.SECURITY_EVENT_MAX_MEMORY || '10000', 10);
|
|
const MAX_EVENTS_ON_DISK = parseInt(process.env.SECURITY_EVENT_MAX_DISK || '100000', 10);
|
|
|
|
// DC-116: size trigger for disk trim. Env-overridable so operators (and tests)
|
|
// can tighten it without rebuilding. Default unchanged: 50MB.
|
|
const TRIM_TARGET_FACTOR = 0.8; // post-trim target: ≤80% of the byte budget
|
|
const DEFAULT_TRIM_SIZE_LIMIT = parseInt(
|
|
process.env.SECURITY_EVENT_TRIM_BYTES || String(50 * 1024 * 1024), 10);
|
|
|
|
const VALID_SOURCE_TYPES = new Set(['api', 'caddy', 'fail2ban', 'shared-bans', 'syslog', 'agent']);
|
|
const VALID_SEVERITIES = new Set(['info', 'notice', 'warn', 'error', 'critical']);
|
|
const VALID_OUTCOMES = new Set(['success', 'failure', 'denied', 'blocked', 'rate-limited', 'error', 'unknown']);
|
|
|
|
class SecurityEventStore extends EventEmitter {
|
|
constructor(opts = {}) {
|
|
super();
|
|
this.filePath = opts.filePath || EVENT_STORE_FILE;
|
|
this.maxMemory = opts.maxMemory || MAX_EVENTS_IN_MEMORY;
|
|
this.maxDisk = opts.maxDisk || MAX_EVENTS_ON_DISK;
|
|
// DC-116: byte budget for the on-disk file (trigger AND target of _trim)
|
|
this.trimSizeLimit = opts.trimSizeLimit != null ? opts.trimSizeLimit : DEFAULT_TRIM_SIZE_LIMIT;
|
|
this.log = opts.log || console;
|
|
this.events = []; // newest first
|
|
this.byId = new Map();
|
|
this.lastWriteLine = 0; // byte offset of last successfully-written line
|
|
this.writeQueue = []; // serialized write buffer
|
|
this.writing = false;
|
|
this._trimScheduled = false; // DC-116: one in-flight trim at a time
|
|
this._load();
|
|
}
|
|
|
|
/**
|
|
* Load existing events from disk into memory (newest first).
|
|
* Skips corrupt lines — logs warning but continues.
|
|
*/
|
|
_load() {
|
|
try {
|
|
if (!fs.existsSync(this.filePath)) {
|
|
this.log.info?.('security', 'event store: starting fresh (no file)', { path: this.filePath });
|
|
return;
|
|
}
|
|
const content = fs.readFileSync(this.filePath, 'utf8');
|
|
const lines = content.split('\n').filter(l => l.trim());
|
|
this.log.info?.('security', 'event store: loading from disk', { total_lines: lines.length, path: this.filePath });
|
|
|
|
let loaded = 0;
|
|
let skipped = 0;
|
|
// Walk from end backwards so newest are first
|
|
for (let i = lines.length - 1; i >= 0 && loaded < this.maxMemory; i--) {
|
|
const line = lines[i].trim();
|
|
if (!line) continue;
|
|
try {
|
|
const ev = JSON.parse(line);
|
|
if (!this._isValidShape(ev)) {
|
|
skipped++;
|
|
continue;
|
|
}
|
|
this.events.push(ev);
|
|
this.byId.set(ev.id, ev);
|
|
loaded++;
|
|
} catch (e) {
|
|
skipped++;
|
|
// Don't log every line — could be thousands of corrupted lines
|
|
}
|
|
}
|
|
this.log.info?.('security', 'event store: load complete', { loaded, skipped, kept_in_memory: this.events.length });
|
|
} catch (e) {
|
|
this.log.error?.('security', 'event store: load failed', { error: e.message });
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Validate minimum shape of an event before storing/returning it.
|
|
* Permissive — accepts extras, just requires the spine.
|
|
*/
|
|
_isValidShape(ev) {
|
|
if (!ev || typeof ev !== 'object') return false;
|
|
if (typeof ev.id !== 'string' || !ev.id) return false;
|
|
if (typeof ev.ts !== 'string' || !ev.ts) return false;
|
|
if (typeof ev.source_host !== 'string') return false;
|
|
if (typeof ev.source_type !== 'string' || !VALID_SOURCE_TYPES.has(ev.source_type)) return false;
|
|
return true;
|
|
}
|
|
|
|
/**
|
|
* Append a new event. Validates shape, writes to disk, indexes in memory,
|
|
* emits 'event' for live-tail subscribers.
|
|
*
|
|
* @param {object} partial - event without id (one will be generated)
|
|
* @returns {object} the stored event
|
|
*/
|
|
append(partial) {
|
|
const event = this._normalize(partial);
|
|
const validationError = this._validate(event);
|
|
if (validationError) {
|
|
this.log.warn?.('security', 'rejected invalid event', { error: validationError, partial });
|
|
throw new Error(`Invalid event: ${validationError}`);
|
|
}
|
|
|
|
// Write to disk first (durability), then index in memory.
|
|
// We queue the write so multiple append() calls don't interleave on the same fd.
|
|
this.writeQueue.push(event);
|
|
this._flushQueue();
|
|
|
|
// Index (in-memory only — newest first)
|
|
this.events.unshift(event);
|
|
this.byId.set(event.id, event);
|
|
if (this.events.length > this.maxMemory) {
|
|
const evicted = this.events.pop();
|
|
this.byId.delete(evicted.id);
|
|
}
|
|
|
|
this.emit('event', event);
|
|
return event;
|
|
}
|
|
|
|
/**
|
|
* Normalize partial event — fill in defaults, generate id+ts.
|
|
*/
|
|
_normalize(p) {
|
|
const now = new Date().toISOString();
|
|
return {
|
|
id: p.id || crypto.randomUUID(),
|
|
ts: p.ts || now,
|
|
source_host: p.source_host || 'unknown',
|
|
source_type: p.source_type || 'api',
|
|
actor: p.actor || null, // IP, user, agent_id
|
|
target: p.target || null, // endpoint, service id, host
|
|
action: p.action || 'unknown', // free-form but stable per source_type
|
|
outcome: p.outcome || 'unknown',
|
|
severity: p.severity || 'info',
|
|
message: p.message || null, // human-readable one-liner
|
|
metadata: p.metadata && typeof p.metadata === 'object' ? p.metadata : {},
|
|
...(p.tags && Array.isArray(p.tags) ? { tags: p.tags } : {}),
|
|
};
|
|
}
|
|
|
|
_validate(ev) {
|
|
if (!VALID_SOURCE_TYPES.has(ev.source_type)) return `bad source_type: ${ev.source_type}`;
|
|
if (!VALID_SEVERITIES.has(ev.severity)) return `bad severity: ${ev.severity}`;
|
|
if (!VALID_OUTCOMES.has(ev.outcome)) return `bad outcome: ${ev.outcome}`;
|
|
if (typeof ev.actor === 'string' && ev.actor.length > 256) return 'actor too long';
|
|
if (typeof ev.target === 'string' && ev.target.length > 512) return 'target too long';
|
|
return null;
|
|
}
|
|
|
|
/**
|
|
* Serialize appends to disk. Writes one line at a time, doesn't truncate.
|
|
* Disk trimming happens separately via _maybeTrim() — and only while the
|
|
* write queue is empty, so an append can never land on the unlinked
|
|
* pre-trim inode (DC-116).
|
|
*/
|
|
_flushQueue() {
|
|
if (this.writing) return;
|
|
const next = this.writeQueue.shift();
|
|
if (!next) {
|
|
// Write path idle — safe point to run a pending trim (no in-flight
|
|
// append can race the rename; new appends queue behind this.writing).
|
|
this._maybeTrim();
|
|
return;
|
|
}
|
|
this.writing = true;
|
|
const line = JSON.stringify(next) + '\n';
|
|
fs.appendFile(this.filePath, line, 'utf8', (err) => {
|
|
this.writing = false;
|
|
if (err) {
|
|
this.log.error?.('security', 'write failed', { error: err.message });
|
|
// Re-queue so we don't lose the event on transient errors. Do NOT
|
|
// auto-retry here — a persistent failure (disk full, perms) would
|
|
// turn setImmediate into a hot loop. The next append() re-kicks
|
|
// the flush (same semantics as before DC-116).
|
|
this.writeQueue.unshift(next);
|
|
return;
|
|
}
|
|
// Drain the rest of the queue before considering a trim. (Note: this
|
|
// merely defers to the next tick — it batches, it does not throttle;
|
|
// with an instantly-draining queue each drain can still end in a trim.)
|
|
if (this.writeQueue.length > 0) {
|
|
setImmediate(() => this._flushQueue());
|
|
} else {
|
|
this._maybeTrim();
|
|
}
|
|
});
|
|
}
|
|
|
|
/**
|
|
* Trim disk log when it exceeds the byte budget. Runs on the write path
|
|
* only when the queue is empty (see _flushQueue), and at most one trim is
|
|
* in flight at a time. Strategy: rewrite the file keeping the most recent
|
|
* maxDisk lines, atomically (write tmp + rename).
|
|
*
|
|
* DC-116 fix — two prior bugs:
|
|
* 1. Trigger/curer mismatch: the trigger was size-based (>50MB) but the
|
|
* curer was line-count-based (keep maxDisk=100k lines). If the average
|
|
* line exceeds SIZE_LIMIT/maxDisk (~524B) the trim no-ops forever while
|
|
* the size trigger keeps firing — unbounded file + a full-file stat
|
|
* (and potentially re-read) on every append. Now: trim fires on size
|
|
* and keeps the most recent maxDisk LINES OR enough BYTES to get under
|
|
* 80% of the budget, whichever retains fewer lines — the file always
|
|
* shrinks back below the trigger.
|
|
* 2. Trim/append race: trim renamed over the file while unrelated appends
|
|
* were in flight, silently losing them to the unlinked inode. Now trim
|
|
* runs only between writes (queue empty, this.writing false) and holds
|
|
* the write lock for its duration.
|
|
*/
|
|
_maybeTrim() {
|
|
if (this._trimScheduled) return; // one at a time
|
|
fs.stat(this.filePath, (err, st) => {
|
|
if (err || !st) return;
|
|
if (st.size < this.trimSizeLimit) return;
|
|
this._trimScheduled = true;
|
|
this.writing = true; // hold the write lock for the whole trim
|
|
this._trim(st.size);
|
|
});
|
|
}
|
|
|
|
_trim(fileSizeBytes) {
|
|
this.log.info?.('security', 'trimming event store', {
|
|
file: this.filePath,
|
|
size_bytes: fileSizeBytes,
|
|
keep_lines: this.maxDisk,
|
|
budget_bytes: this.trimSizeLimit,
|
|
});
|
|
fs.readFile(this.filePath, 'utf8', (err, content) => {
|
|
const done = (e) => {
|
|
this._trimScheduled = false;
|
|
this.writing = false; // release the write lock
|
|
if (this.writeQueue.length > 0) setImmediate(() => this._flushQueue());
|
|
if (e) this.log.error?.('security', 'trim failed', { error: e.message });
|
|
};
|
|
if (err) return done(err);
|
|
|
|
const lines = content.split('\n').filter(l => l.trim());
|
|
if (lines.length === 0) return done();
|
|
|
|
// Byte-aware reconciliation: keep the most recent maxDisk lines, but if
|
|
// that slice alone still exceeds ~80% of the byte budget, drop further
|
|
// lines (oldest first) until it fits. Always retains at least one line.
|
|
const keep = lines.slice(-this.maxDisk);
|
|
let keepBytes = Buffer.byteLength(keep.join('\n') + '\n', 'utf8');
|
|
const byteCeiling = Math.floor(this.trimSizeLimit * TRIM_TARGET_FACTOR);
|
|
while (keep.length > 1 && keepBytes > byteCeiling) {
|
|
keepBytes -= Buffer.byteLength(keep.shift() + '\n', 'utf8');
|
|
}
|
|
|
|
const kept = keep.join('\n') + '\n';
|
|
const tmp = this.filePath + '.tmp';
|
|
fs.writeFile(tmp, kept, 'utf8', (e) => {
|
|
if (e) return done(e);
|
|
fs.rename(tmp, this.filePath, done);
|
|
});
|
|
});
|
|
}
|
|
|
|
/**
|
|
* Query events. All filters are AND-combined. Results are newest-first.
|
|
*
|
|
* `total` is the true count of ALL matching events in the memory window
|
|
* (DC-116 fix: the loop previously broke at offset+limit, so `total` was
|
|
* silently capped at the page size — the dashboard's "N events (24h)" stat
|
|
* and hosts/:id/health events_24h read 1000 when the real count was tens
|
|
* of thousands). The scan now always completes; per-page cost is bounded
|
|
* by the in-memory cap (maxMemory, default 10k).
|
|
*
|
|
* @param {object} q - query
|
|
* limit : number, default 100, max 1000
|
|
* offset : number, default 0
|
|
* source_type: string or array
|
|
* source_host: string
|
|
* severity : string or array
|
|
* outcome : string or array
|
|
* actor : string (exact or prefix match with `actor_prefix`)
|
|
* action : string
|
|
* since : ISO timestamp (inclusive)
|
|
* until : ISO timestamp (exclusive)
|
|
*/
|
|
query(q = {}) {
|
|
const limit = Math.min(parseInt(q.limit || '100', 10), 1000);
|
|
const offset = parseInt(q.offset || '0', 10);
|
|
const match = this.compileFilter(q);
|
|
|
|
let total = 0;
|
|
const page = [];
|
|
for (const ev of this.events) {
|
|
if (!match(ev)) continue;
|
|
total++;
|
|
if (total > offset && page.length < limit) page.push(ev);
|
|
}
|
|
|
|
return {
|
|
total,
|
|
events: page,
|
|
};
|
|
}
|
|
|
|
/**
|
|
* Compile the query filters into a single predicate. Shared by query()
|
|
* (paged access) and filterEvents() (full-set access) so the two can
|
|
* never drift on filter semantics (DC-120).
|
|
*
|
|
* All filters AND-combine; an absent filter matches everything.
|
|
*/
|
|
compileFilter(q = {}) {
|
|
const sourceTypes = this._toArr(q.source_type);
|
|
const severities = this._toArr(q.severity);
|
|
const outcomes = this._toArr(q.outcome);
|
|
return (ev) => {
|
|
if (sourceTypes.length && !sourceTypes.includes(ev.source_type)) return false;
|
|
if (q.source_host && ev.source_host !== q.source_host) return false;
|
|
if (severities.length && !severities.includes(ev.severity)) return false;
|
|
if (outcomes.length && !outcomes.includes(ev.outcome)) return false;
|
|
if (q.actor && ev.actor !== q.actor) return false;
|
|
if (q.actor_prefix && (!ev.actor || !ev.actor.startsWith(q.actor_prefix))) return false;
|
|
if (q.action && ev.action !== q.action) return false;
|
|
if (q.since && ev.ts < q.since) return false;
|
|
if (q.until && ev.ts >= q.until) return false;
|
|
if (q.target && ev.target !== q.target) return false;
|
|
return true;
|
|
};
|
|
}
|
|
|
|
/**
|
|
* DC-120: full filtered set, newest-first, for route-level aggregation
|
|
* that the paged query() API can't express (e.g. per-IP outcome
|
|
* breakdowns across every caddy event in a window — query() pages at
|
|
* 1000 and `total` alone can't rebuild the per-key maps).
|
|
*
|
|
* Cost is bounded by the in-memory cap (maxMemory, default 10k) — the
|
|
* same bound query() already scans — so callers cannot request
|
|
* unbounded work. Filters are identical to query() by construction
|
|
* (shared compileFilter).
|
|
*/
|
|
filterEvents(q = {}) {
|
|
const match = this.compileFilter(q);
|
|
return this.events.filter(match);
|
|
}
|
|
|
|
_toArr(v) {
|
|
if (!v) return [];
|
|
if (Array.isArray(v)) return v;
|
|
return String(v).split(',').map(s => s.trim()).filter(Boolean);
|
|
}
|
|
|
|
/**
|
|
* Compute aggregates for dashboards. Returns counts + top-N per dimension.
|
|
* Cheap because events array is bounded by maxMemory.
|
|
*/
|
|
stats(opts = {}) {
|
|
const sinceMs = opts.since ? Date.parse(opts.since) : null;
|
|
const filtered = sinceMs
|
|
? this.events.filter(e => Date.parse(e.ts) >= sinceMs)
|
|
: this.events;
|
|
|
|
const bySeverity = {};
|
|
const byOutcome = {};
|
|
const bySource = {};
|
|
const byHost = {};
|
|
const byAction = {};
|
|
const actorCount = {};
|
|
const targetCount = {};
|
|
|
|
for (const ev of filtered) {
|
|
bySeverity[ev.severity] = (bySeverity[ev.severity] || 0) + 1;
|
|
byOutcome[ev.outcome] = (byOutcome[ev.outcome] || 0) + 1;
|
|
bySource[ev.source_type] = (bySource[ev.source_type] || 0) + 1;
|
|
byHost[ev.source_host] = (byHost[ev.source_host] || 0) + 1;
|
|
byAction[ev.action] = (byAction[ev.action] || 0) + 1;
|
|
if (ev.actor) actorCount[ev.actor] = (actorCount[ev.actor] || 0) + 1;
|
|
if (ev.target) targetCount[ev.target] = (targetCount[ev.target] || 0) + 1;
|
|
}
|
|
|
|
return {
|
|
window: { since: opts.since || null, count: filtered.length },
|
|
by_severity: bySeverity,
|
|
by_outcome: byOutcome,
|
|
by_source: bySource,
|
|
by_host: byHost,
|
|
top_actions: this._topN(byAction, 10),
|
|
top_actors: this._topN(actorCount, 10),
|
|
top_targets: this._topN(targetCount, 10),
|
|
};
|
|
}
|
|
|
|
_topN(obj, n) {
|
|
return Object.entries(obj)
|
|
.sort((a, b) => b[1] - a[1])
|
|
.slice(0, n)
|
|
.map(([key, count]) => ({ key, count }));
|
|
}
|
|
|
|
/**
|
|
* Get one event by id, or null.
|
|
*/
|
|
get(id) {
|
|
return this.byId.get(id) || null;
|
|
}
|
|
|
|
/**
|
|
* Total events currently in memory.
|
|
*/
|
|
size() {
|
|
return this.events.length;
|
|
}
|
|
}
|
|
|
|
// Singleton accessor. Routes call this; tests can construct their own.
|
|
let _instance = null;
|
|
function getStore(opts) {
|
|
if (_instance) return _instance;
|
|
_instance = new SecurityEventStore(opts);
|
|
return _instance;
|
|
}
|
|
|
|
module.exports = { SecurityEventStore, getStore, VALID_SOURCE_TYPES, VALID_SEVERITIES, VALID_OUTCOMES }; |