/** * Security Event Workers * * Background processes that watch external sources for security events and * push them into the unified security event store: * * 1. Caddy access log tail — parses /var/log/caddy/access.log (JSON format) * and emits one event per request. Severity escalates for 4xx/5xx and * credential-endpoint hits. * * 2. shared_bans apply tail — parses /var/log/shared-bans-apply.log for * IP-blocklist changes. Emits 'info' events so the dashboard timeline * shows when IPs were banned/promoted. * * 3. fail2ban log tail — parses /var/log/fail2ban.log for ban/unban * actions. SSH jail is the default; can extend to other jails. * * Each worker: * - Starts on app boot (via server.js) * - Tracks its byte offset in the log file so it survives restarts (no re-emit) * - Auto-recovers from truncated/rotated log files * - Has its own error handling — one worker dying doesn't take down the others * * To use the Caddy worker, configure Caddy to log in JSON format: * * { * log default { * output file /var/log/caddy/access.log { * roll_size 100mb * roll_keep 10 * } * format json * } * } * * Then drop a fail2ban jail for HTTP 401/403 patterns — see HARDENING.md P1.1. */ const fs = require('fs'); const path = require('path'); const os = require('os'); const platformPaths = require('../../platform-paths'); const { getStore } = require('./event-store'); const HOSTNAME = os.hostname(); /** * Generic tail-follower with offset persistence. * Watches `filePath`, emits each new line via `onLine(line)`. * Persists last-read offset to `stateFile` so restarts don't re-process. * On file truncation (rotation), resets offset to 0. */ function createTail({ filePath, stateFile, onLine, label = 'tail', pollMs = 1000 }) { let offset = 0; let buffer = ''; let stopped = false; // Load persisted offset try { if (fs.existsSync(stateFile)) { offset = parseInt(fs.readFileSync(stateFile, 'utf8').trim(), 10) || 0; } } catch {} function persistOffset() { try { fs.writeFileSync(stateFile, String(offset), 'utf8'); } catch {} } function tick() { if (stopped) return; fs.stat(filePath, (err, st) => { if (err) { // File doesn't exist yet — just wait return setTimeout(tick, pollMs * 5); } // Detect truncation/rotation if (st.size < offset) { offset = 0; buffer = ''; } if (st.size === offset) { return setTimeout(tick, pollMs); } // Read just the new bytes const stream = fs.createReadStream(filePath, { start: offset, end: st.size - 1, encoding: 'utf8', }); stream.on('data', (chunk) => { buffer += chunk; const lines = buffer.split('\n'); buffer = lines.pop() || ''; // last partial stays for (const line of lines) { if (line.trim()) { try { onLine(line); } catch (e) { console.error(`[${label}] onLine threw:`, e.message); } } } }); stream.on('end', () => { offset = st.size; persistOffset(); setTimeout(tick, pollMs); }); stream.on('error', (e) => { console.error(`[${label}] read error:`, e.message); setTimeout(tick, pollMs * 5); }); }); } setTimeout(tick, pollMs); // initial delay so app has finished starting return { stop() { stopped = true; }, getOffset() { return offset; }, }; } /** * Worker 1 — Caddy access log. * Caddy emits JSON per request like: * {"ts":1700000000,"request":{"remote_ip":"1.2.3.4","method":"GET","uri":"/x"},"status":200,...} * We turn that into a security event. */ function startCaddyWorker({ log } = {}) { const caddyLog = process.env.CADDY_ACCESS_LOG || '/var/log/caddy/access.log'; const stateFile = path.join(platformPaths.dataDir, '.caddy-tail-offset'); const store = getStore({ log }); return createTail({ filePath: caddyLog, stateFile, label: 'caddy', onLine: (line) => { let entry; try { entry = JSON.parse(line); } catch { return; } // skip non-JSON lines (Caddy may mix formats) const req = entry.request || {}; const status = entry.status || 0; const ip = req.remote_ip; const method = req.method; const uri = req.uri || ''; const userAgent = (req.headers && req.headers['User-Agent']) || null; // Severity mapping let severity = 'info'; let outcome = 'success'; if (status === 401 || status === 403) { severity = 'warn'; outcome = 'denied'; } else if (status === 429) { severity = 'notice'; outcome = 'rate-limited'; } else if (status >= 500) { severity = 'error'; outcome = 'error'; } else if (status >= 400) { severity = 'notice'; outcome = 'denied'; } // Escalate credential-endpoint hits const sensitivePaths = ['/api/v1/auth/', '/api/v1/totp/', '/api/v1/license/', '/api/v1/credentials/']; if (sensitivePaths.some(p => uri.startsWith(p)) && status >= 400) { severity = 'warn'; } store.append({ source_host: HOSTNAME, source_type: 'caddy', actor: ip, target: `${method} ${uri}`, action: `http.${status}`, outcome, severity, message: `${ip} ${method} ${uri} -> ${status}`, metadata: { status, duration_ms: entry.duration || null, user_agent: userAgent, size: entry.size || null, proto: req.proto || null, }, }); }, }); } /** * Worker 2 — shared_bans apply log. * Already a structured human-readable log: * "2026-07-13 01:35:55 Excluded 6 private/loopback/CGNAT entries from ban list" * "2026-07-13 01:35:56 Applied: 19412 entries in shared_bans" * We emit one event per "Applied" line. Low volume (1 per 5 min) so very cheap. */ function startSharedBansWorker({ log } = {}) { const sbLog = process.env.SHARED_BANS_LOG || '/var/log/shared-bans-apply.log'; const stateFile = path.join(platformPaths.dataDir, '.sb-tail-offset'); const store = getStore({ log }); const APPLIED_RE = /Applied:\s+(\d+)\s+entries/; return createTail({ filePath: sbLog, stateFile, label: 'shared-bans', pollMs: 5000, onLine: (line) => { const m = line.match(APPLIED_RE); if (!m) return; // skip the "Excluded" / "Merged" / "Restored" noise const count = parseInt(m[1], 10); store.append({ source_host: HOSTNAME, source_type: 'shared-bans', actor: 'shared-bans-updater', target: 'shared_bans ipset', action: 'ipset.apply', outcome: 'success', severity: 'info', message: `Applied ${count} entries to shared_bans ipset`, metadata: { count }, }); }, }); } /** * Worker 3 — fail2ban log. * "2026-06-15T21:05:41Z fail2ban.actions [sshd] Ban 1.2.3.4" * "2026-06-15T21:05:41Z fail2ban.actions [sshd] Unban 1.2.3.4" * Emit one event per Ban/Unban. Watched on top of shared_bans because fail2ban * bans are SHORTER-lived (24h default) than shared_bans. */ function startFail2banWorker({ log } = {}) { const f2bLog = process.env.FAIL2BAN_LOG || '/var/log/fail2ban.log'; const stateFile = path.join(platformPaths.dataDir, '.f2b-tail-offset'); const store = getStore({ log }); // Match ISO timestamps followed by [jail] Ban/Unban IP const BAN_RE = /^(\S+).*?\]\s+(Ban|Unban)\s+(\S+)/; return createTail({ filePath: f2bLog, stateFile, label: 'fail2ban', pollMs: 2000, onLine: (line) => { const m = line.match(BAN_RE); if (!m) return; const [, ts, action, ip] = m; const isBan = action === 'Ban'; store.append({ source_host: HOSTNAME, source_type: 'fail2ban', actor: ip, target: 'sshd (or other jail)', action: isBan ? 'ban' : 'unban', outcome: 'success', severity: isBan ? 'notice' : 'info', message: `${action} ${ip}`, metadata: { ts, source: 'fail2ban', }, }); }, }); } /** * Start all workers. Returns a stop function that shuts them all down. */ function startAll({ log } = {}) { const workers = []; try { workers.push(startCaddyWorker({ log })); } catch (e) { console.error('[workers] caddy worker failed to start:', e.message); } try { workers.push(startSharedBansWorker({ log })); } catch (e) { console.error('[workers] shared_bans worker failed to start:', e.message); } try { workers.push(startFail2banWorker({ log })); } catch (e) { console.error('[workers] fail2ban worker failed to start:', e.message); } return { stop() { workers.forEach(w => { try { w.stop(); } catch {} }); }, workers, }; } module.exports = { createTail, startCaddyWorker, startSharedBansWorker, startFail2banWorker, startAll, };