Monitoreo de directorios y escaneo antivirus en tiempo real con Rust y ClamAV

Kevin van Zonneveld

Cofundador · Ámsterdam, Países Bajos · Mostrar bio

Crea un monitor en Rust que escanee archivos completos con un demonio local de ClamAV. Informa resultados limpios, infectados y errores del escáner sin mover ni eliminar archivos. Solo una respuesta explícita de resultado limpio cuenta como tal.

¿Por qué usar monitoreo en tiempo real y Rust?

Los eventos del sistema de archivos pueden llegar antes de que termine la escritura. Una espera o un tamaño de archivo sin cambios no demuestran que haya finalizado. Por eso, este ejemplo usa un contrato con el productor: escribe un archivo con nombre único terminado en .part, ciérralo y luego renómbralo de forma atómica a .ready en el mismo directorio. Nunca modifiques un archivo .ready. Mantén el directorio privado para la aplicación y su productor de confianza.

El monitor escanea solo archivos .ready del nivel superior. Es una herramienta de notificación, no una barrera de control de acceso: un resultado describe los bytes escaneados y no autoriza una lectura posterior de una ruta mutable.

Requisitos previos

Este tutorial usa Linux y Bash, con Rust y Cargo 1.98.1 y ClamAV 1.5.4. Prepara un demonio local clamd con una base de datos de firmas actualizada y un socket Unix accesible para el proceso de Rust. Los comandos siguientes están destinados a Linux; el cliente usa sockets Unix.

Configura el demonio de ClamAV (clamd)

En Debian o Ubuntu, la guía de paquetes de ClamAV⁠ enumera los paquetes del demonio y del actualizador de definiciones:

sudo apt-get update &&
sudo apt-get install clamav-daemon clamav-freshclam

En el archivo clamd.conf de la distribución, conserva la configuración de la base de datos y del usuario del servicio, y configura un socket local. Este es un fragmento de configuración; asegúrate de que no quede habilitada ninguna directiva TCPSocket:

LocalSocket /var/run/clamav/clamd.ctl
LocalSocketMode 660
StreamMaxLength 10M
MaxFileSize 10M
MaxScanSize 20M
AlertExceedsMax yes

Concede al usuario del servicio de la aplicación acceso al grupo del socket y luego reinicia el servicio. Deja que el servicio FreshClam de la distribución mantenga actualizada la base de datos. No ejecutes un segundo actualizador manual sobre su base de datos bloqueada.

sudo systemctl enable --now clamav-freshclam &&
sudo systemctl restart clamav-daemon

La documentación del protocolo de ClamAV⁠ describe el socket y la estructura de las tramas de INSTREAM. Clamd no tiene autenticación TCP⁠; este ejemplo usa únicamente un socket local con acceso controlado mediante permisos. Configura los límites de archivos contenedores y de escaneo según tu carga de trabajo. Si AlertExceedsMax está habilitado, las alertas del demonio por límites excedidos⁠ no deben tratarse como resultados limpios.

Configura el proyecto

Empieza en un directorio con permiso de escritura fuera de un proyecto Cargo existente. Cargo puede agregar un paquete nuevo a un espacio de trabajo antecesor⁠, y la configuración se hereda de los directorios superiores⁠. Este bloque de Bash rechaza esas ubicaciones antes de crear nada. Solo entra en el proyecto nuevo cuando cargo new se ejecuta correctamente; si ya existe un directorio realtime_virus_scanner, lo deja intacto.

(
    cd -P . || exit 1
    scanner_parent=$PWD
    while :; do
        if [ -e "$scanner_parent/Cargo.toml" ] ||
           [ -e "$scanner_parent/.cargo/config.toml" ] ||
           [ -e "$scanner_parent/.cargo/config" ]; then
            printf 'Choose a directory outside existing Cargo projects/configuration.\n' >&2
            exit 1
        fi
        [ "$scanner_parent" = / ] && break
        scanner_parent=${scanner_parent%/*}
        [ -n "$scanner_parent" ] || scanner_parent=/
    done
    cargo new realtime_virus_scanner --edition 2021 --vcs none
) && cd realtime_virus_scanner

Una vez completada la configuración, reemplaza el archivo Cargo.toml del nuevo proyecto por lo siguiente:

[package]
name = "realtime_virus_scanner"
version = "0.1.0"
edition = "2021"

[dependencies]
clap = { version = "=4.5.50", features = ["derive"] }
notify = "=8.2.0"
tokio = { version = "=1.48.0", features = ["fs", "io-util", "macros", "net", "rt-multi-thread", "signal", "sync", "time"] }

Conserva Cargo.lock en tu aplicación. El ejemplo implementa el pequeño intercambio INSTREAM directamente con Tokio, en lugar de depender de una biblioteca cliente adicional para ClamAV.

Interfaz de línea de comandos con clap

La CLI acepta --directory y --socket. Agrega --once para escanear los archivos completos existentes y salir: cero significa que todos los archivos seleccionados obtuvieron un resultado limpio; uno indica que se produjo una infección o un error del escáner. Una selección vacía también termina con cero; eso no indica nada sobre los archivos .part sin terminar. El modo de monitoreo repite los escaneos hasta que presionas Ctrl+C e informa cada resultado.

Monitoreo de directorios en tiempo real con notify

Un observador recomendado⁠ activa el escáner después de un evento de creación, modificación o eliminación. Un canal con capacidad para un elemento agrupa los eventos porque cada activación vuelve a escanear el directorio. Un intervalo de reconciliación de cinco segundos también permite encontrar archivos cuando se pierden eventos. Los eventos de lectura o acceso se ignoran para evitar que los escaneos se activen a sí mismos.

Escaneo antivirus asíncrono con Tokio

Cada escaneo toma una instantánea de tamaño limitado, la transmite en bloques de 64 KiB y lee como máximo 4097 bytes de respuesta. Solo se acepta la respuesta exacta, única y terminada que indica un resultado limpio. Las respuestas vacías, truncadas, demasiado grandes, múltiples o no reconocidas se rechazan por seguridad. Una detección de infección o de límite del motor no constituye un resultado limpio. Un plazo máximo de diez segundos abarca la lectura del archivo y la comunicación con clamd.

Integra todo: src/main.rs

Guarda este programa completo como src/main.rs:

use clap::Parser;
use notify::{Event, EventKind, RecursiveMode, Watcher};
use std::{error::Error, path::{Path, PathBuf}, time::Duration};
use tokio::{
    fs::{self, File},
    io::{AsyncReadExt, AsyncWriteExt},
    net::UnixStream,
    sync::mpsc,
    time::{interval, timeout},
};

type AppResult<T> = Result<T, Box<dyn Error + Send + Sync>>;
const MAX_BYTES: u64 = 10 * 1024 * 1024;

#[derive(Parser)]
struct Args {
    #[arg(long)]
    directory: PathBuf,
    #[arg(long, default_value = "/var/run/clamav/clamd.ctl")]
    socket: PathBuf,
    #[arg(long)]
    once: bool,
}

#[derive(Debug, PartialEq)]
enum Verdict {
    Clean,
    Infected,
    ScannerError,
}

fn parse_reply(reply: &[u8]) -> Verdict {
    if reply == b"stream: OK\0" {
        return Verdict::Clean;
    }
    if let Some(name) = reply.strip_prefix(b"stream: ")
        .and_then(|value| value.strip_suffix(b" FOUND\0"))
    {
        if !name.is_empty() && !name.iter().any(|byte| byte.is_ascii_control()) {
            return Verdict::Infected;
        }
    }
    Verdict::ScannerError
}

async fn scan(path: &Path, socket: &Path) -> AppResult<Verdict> {
    // The private-directory producer contract forbids modifying published .ready files.
    if !fs::symlink_metadata(path).await?.file_type().is_file() {
        return Err("Input must be a regular, non-symlink file".into());
    }
    let file = File::open(path).await?;
    if !file.metadata().await?.is_file() {
        return Err("Input must be a regular file".into());
    }
    let mut bytes = Vec::new();
    file.take(MAX_BYTES + 1).read_to_end(&mut bytes).await?;
    if bytes.len() as u64 > MAX_BYTES {
        return Err("Input exceeds 10 MiB".into());
    }

    let mut stream = UnixStream::connect(socket).await?;
    stream.write_all(b"zINSTREAM\0").await?;
    for chunk in bytes.chunks(64 * 1024) {
        stream.write_all(&(chunk.len() as u32).to_be_bytes()).await?;
        stream.write_all(chunk).await?;
    }
    stream.write_all(&0u32.to_be_bytes()).await?;
    let mut reply = Vec::new();
    stream.take(4097).read_to_end(&mut reply).await?;
    Ok(if reply.len() > 4096 {
        Verdict::ScannerError
    } else {
        parse_reply(&reply)
    })
}

async fn scan_directory(args: &Args) -> AppResult<bool> {
    let mut entries = fs::read_dir(&args.directory).await?;
    let mut all_clean = true;
    while let Some(entry) = entries.next_entry().await? {
        let path = entry.path();
        if path.extension().and_then(|value| value.to_str()) != Some("ready") {
            continue;
        }
        let verdict = match timeout(Duration::from_secs(10), scan(&path, &args.socket)).await {
            Ok(Ok(verdict)) => verdict,
            _ => Verdict::ScannerError,
        };
        all_clean &= verdict == Verdict::Clean;
        // Debug path formatting escapes newlines and control characters in filenames.
        println!("{verdict:?} {:?}", path.file_name());
    }
    Ok(all_clean)
}

async fn monitor(args: &Args) -> AppResult<bool> {
    if args.once {
        return scan_directory(args).await;
    }
    let (tx, mut rx) = mpsc::channel(1);
    let mut watcher = notify::recommended_watcher(move |event: notify::Result<Event>| {
        let changed = match event {
            Ok(event) => matches!(
                event.kind,
                EventKind::Create(_) | EventKind::Modify(_) | EventKind::Remove(_)
            ),
            Err(_) => {
                eprintln!("Watcher error; periodic reconciliation remains active.");
                true
            }
        };
        if changed {
            // A full queue already guarantees a complete directory reconciliation.
            let _ = tx.try_send(());
        }
    })?;
    watcher.watch(&args.directory, RecursiveMode::NonRecursive)?;
    let mut ticks = interval(Duration::from_secs(5));
    loop {
        tokio::select! {
            _ = ticks.tick() => {}
            event = rx.recv() => {
                if event.is_none() {
                    return Err("Watcher channel closed".into());
                }
            }
        }
        scan_directory(args).await?;
    }
}

async fn run() -> AppResult<bool> {
    let args = Args::parse();
    tokio::select! {
        result = monitor(&args) => result,
        result = tokio::signal::ctrl_c() => {
            result?;
            Err("Scanning interrupted".into())
        }
    }
}

#[tokio::main]
async fn main() {
    match run().await {
        Ok(true) => {}
        Ok(false) => std::process::exit(1),
        Err(_) => {
            eprintln!("Scanner stopped; check the directory, socket, and daemon configuration.");
            std::process::exit(1);
        }
    }
}

Desde el directorio del proyecto, compila e inicia el monitor con una nueva bandeja de entrada privada. Cada paso debe completarse correctamente antes de ejecutar el siguiente, para que una compilación fallida no pueda iniciar un ejecutable antiguo:

mkdir -m 700 inbox &&
cargo build --jobs 2 --target-dir target &&
./target/debug/realtime_virus_scanner --directory inbox --socket /var/run/clamav/clamd.ctl

Cambia el argumento del socket si tu demonio usa otra ruta. Deja esta terminal en ejecución. En otra terminal, desde el directorio del proyecto, publica un archivo terminado usando un par de nombres que no estén en uso:

(
    set -o noclobber
    test ! -e inbox/example.ready &&
    test ! -L inbox/example.ready &&
    test ! -e inbox/example.part &&
    test ! -L inbox/example.part &&
    printf 'A completed example file.\n' > inbox/example.part &&
    mv -n -- inbox/example.part inbox/example.ready &&
    test ! -e inbox/example.part
)

Un escaneo limpio imprime Clean Some("example.ready"), posiblemente más de una vez porque las reconciliaciones vuelven a escanear los archivos completos. El bloque del productor devuelve un código distinto de cero si cualquiera de los nombres ya existe, sin reemplazar su contenido. Usa nombres nuevos para los archivos posteriores y no ejecutes productores simultáneamente con los mismos nombres. El comando printf cierra el archivo antes de que mv lo publique.

Presiona Ctrl+C en la terminal del monitor para detenerlo. Para escanear la misma bandeja de entrada una sola vez, ejecuta:

./target/debug/realtime_virus_scanner --directory inbox --socket /var/run/clamav/clamd.ctl --once

Gestión de errores y notificaciones

ScannerError significa que el escaneo falló, lo que incluye un socket no disponible, una entrada de más de 10 MiB, un enlace simbólico o el vencimiento del plazo. Infected también abarca las alertas por límites excedidos configuradas en ClamAV. Ninguno de estos resultados debe permitir que un archivo pase a los consumidores posteriores. El ejemplo deja deliberadamente todos los archivos en su lugar para que un fallo del escaneo no pueda eliminar una subida.

Ctrl+C descarta el futuro del escaneo activo, cierra su socket, libera el observador y sale con un código distinto de cero. En el modo de monitoreo, un escaneo fallido se reintenta durante una reconciliación posterior. Usa --once cuando quien lo invoca necesite un código de salida de error para un lote que no haya obtenido un resultado limpio.

Consideraciones de rendimiento

Hay un solo escaneo activo a la vez, una activación pendiente, una instantánea de tamaño limitado por archivo y una respuesta de tamaño limitado. La enumeración del directorio no acumula una cola de rutas. Volver a escanear todos los archivos completos sacrifica rendimiento a cambio de un ejemplo pequeño y recuperable; un servicio más grande debería llevar un registro persistente del trabajo y asociar cada veredicto con el objeto inmutable exacto que escaneó.

Este monitor no es recursivo, y el intervalo de cinco segundos es una programación de reconciliación, no un plazo de detección garantizado. Los lotes grandes y los escaneos lentos tardan más. Mantén las firmas y prueba los límites de archivos contenedores de tu demonio como parte de la operación del servicio.

Conclusión

Solo los productores de confianza deberían publicar archivos .ready, y los consumidores no deben interpretar este monitor como un permiso para servirlos. Clean significa que el motor de ClamAV configurado no encontró ninguna detección en los bytes que escaneó, no que el archivo sea inofensivo. Mantén actualizadas las firmas y elige límites del demonio adecuados para los formatos que aceptas.

Transloadit también usa ClamAV para filtrar archivos mediante su Robot 🤖 /file/virusscan.