dashcaddy/dashcaddy-api/src/security/event-workers.js

/**
 * 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 { 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(process.env.DATA_DIR || path.join(__dirname, '../../data'), '.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(process.env.DATA_DIR || path.join(__dirname, '../../data'), '.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(process.env.DATA_DIR || path.join(__dirname, '../../data'), '.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,
};