prometheus/service/clamd.js

/**
 * A minimal client for clamd — ClamAV's long-running daemon.
 *
 * We speak the INSTREAM command over TCP. The protocol is small enough that
 * reaching for a third-party clamd client would obscure what we do more than
 * explain it:
 *
 *   client → server:  "zINSTREAM\0"
 *   client → server:  <4-byte big-endian length><chunk>  (repeated)
 *   client → server:  <4-byte 0>                         (end-of-stream)
 *   client ← server:  "stream: OK\0"                     (clean)
 *                 or  "stream: <SignatureName> FOUND\0"  (infected)
 *                 or  "ERROR: <message>\0"               (operational problem)
 *
 * Reference: https://docs.clamav.net/manual/Usage/Scanning.html#clamd-protocol
 *
 * The `MOCK_SCAN` env var short-circuits all of this for local development on a
 * box without ClamAV installed. Values: `clean`, `infected`, `error`.
 */

import { connect } from 'node:net';
import { PassThrough } from 'node:stream';

const CLAMD_HOST = process.env.CLAMD_HOST || '127.0.0.1';
const CLAMD_PORT = Number(process.env.CLAMD_PORT) || 3310;
const CLAMD_CHUNK = 65536;           // 64 KiB per INSTREAM frame
const CLAMD_SOCKET_TIMEOUT = 60_000; // 60 s per scan — generous for big files

/**
 * Scan a stream through clamd and return the verdict.
 *
 * @param {NodeJS.ReadableStream} source  the file bytes
 * @returns {Promise<{verdict: 'clean'|'infected'|'error', engine: string, engine_version?: string, signatures: string[], raw: string, duration_ms: number}>}
 */
export async function scanStream(source) {
  const mock = process.env.MOCK_SCAN;
  if (mock) {
    return mockScan(mock, source);
  }

  const started = Date.now();

  return new Promise((resolve, reject) => {
    const sock = connect({ host: CLAMD_HOST, port: CLAMD_PORT });
    sock.setTimeout(CLAMD_SOCKET_TIMEOUT);

    let response = Buffer.alloc(0);
    let settled = false;

    const finish = (result) => {
      if (settled) return;
      settled = true;
      sock.destroy();
      result.duration_ms = Date.now() - started;
      resolve(result);
    };

    const fail = (err) => {
      if (settled) return;
      settled = true;
      sock.destroy();
      // Operational problems (daemon down, network error, timeout) don't make
      // the file infected — they make the scan inconclusive. The caller decides
      // what to do with that; we never pretend a scan happened when it didn't.
      resolve({
        verdict: 'error',
        engine: 'clamav',
        signatures: [],
        raw: '',
        error: String(err?.message || err),
        duration_ms: Date.now() - started,
      });
    };

    sock.on('error', fail);
    sock.on('timeout', () => fail(new Error('clamd socket timeout')));

    sock.on('data', (chunk) => {
      response = Buffer.concat([response, chunk]);
      // Response is a single null-terminated string.
      if (response.includes(0)) {
        const text = response.subarray(0, response.indexOf(0)).toString('utf8');
        finish(parseVerdict(text));
      }
    });

    sock.on('close', () => {
      if (!settled) {
        fail(new Error('clamd closed connection without a verdict'));
      }
    });

    sock.on('connect', () => {
      sock.write('zINSTREAM\0');

      // Forward the source through the framing format clamd expects.
      source.on('error', fail);

      source.on('data', (chunk) => {
        if (settled) return;
        const buf = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
        // One source chunk may be bigger than our 64 KiB frame cap; slice it.
        for (let i = 0; i < buf.length; i += CLAMD_CHUNK) {
          const slice = buf.subarray(i, Math.min(i + CLAMD_CHUNK, buf.length));
          const header = Buffer.alloc(4);
          header.writeUInt32BE(slice.length, 0);
          sock.write(header);
          sock.write(slice);
        }
      });

      source.on('end', () => {
        if (settled) return;
        const terminator = Buffer.alloc(4); // 0 means end-of-stream
        sock.write(terminator);
      });
    });
  });
}

/* ---------------------------------------------------------------------- */

/**
 * Translate clamd's response line into a structured verdict.
 */
function parseVerdict(text) {
  const trimmed = text.trim();

  if (/\bOK\b/.test(trimmed) && !/FOUND/i.test(trimmed)) {
    return { verdict: 'clean', engine: 'clamav', signatures: [], raw: trimmed };
  }

  const foundMatch = trimmed.match(/stream:\s*(.+?)\s+FOUND/i);
  if (foundMatch) {
    return {
      verdict: 'infected',
      engine: 'clamav',
      signatures: [foundMatch[1].trim()],
      raw: trimmed,
    };
  }

  // Anything that didn't match the two happy shapes is an operational error.
  return {
    verdict: 'error',
    engine: 'clamav',
    signatures: [],
    raw: trimmed,
    error: trimmed,
  };
}

/**
 * Local-dev short-circuit. Drains the source so the caller's upstream code
 * (SHA computation, byte accounting) still gets the data.
 */
async function mockScan(mode, source) {
  const started = Date.now();
  // Drain the source — the caller's hash/size pipeline needs the bytes to
  // flow through end-to-end even when we aren't really scanning.
  await new Promise((resolve, reject) => {
    const sink = new PassThrough();
    source.on('error', reject);
    sink.on('error', reject);
    sink.on('finish', resolve);
    source.pipe(sink);
  });

  const duration_ms = Date.now() - started;
  if (mode === 'infected') {
    return {
      verdict: 'infected',
      engine: 'mock',
      signatures: ['Mock.Test.Sample'],
      raw: 'MOCK_SCAN=infected',
      duration_ms,
    };
  }
  if (mode === 'error') {
    return {
      verdict: 'error',
      engine: 'mock',
      signatures: [],
      raw: 'MOCK_SCAN=error',
      error: 'MOCK_SCAN=error forced an operational failure',
      duration_ms,
    };
  }
  return {
    verdict: 'clean',
    engine: 'mock',
    signatures: [],
    raw: 'MOCK_SCAN=clean',
    duration_ms,
  };
}

/**
 * One-shot "is clamd alive" check, for the healthcheck endpoint.
 */
export async function clamdPing() {
  if (process.env.MOCK_SCAN) {
    return { ok: true, engine: 'mock' };
  }
  return new Promise((resolve) => {
    const sock = connect({ host: CLAMD_HOST, port: CLAMD_PORT });
    sock.setTimeout(2000);
    let data = '';
    const done = (ok, err = '') => {
      sock.destroy();
      resolve(ok ? { ok: true, engine: 'clamav' } : { ok: false, error: err });
    };
    sock.on('connect', () => sock.write('zPING\0'));
    sock.on('data', (c) => {
      data += c.toString('utf8');
      if (data.includes('PONG')) done(true);
    });
    sock.on('error', (e) => done(false, String(e?.message || e)));
    sock.on('timeout', () => done(false, 'timeout'));
    sock.on('close', () => done(data.includes('PONG')));
  });
}