Monitoramento de diretórios em tempo real e varredura de vírus com Rust e ClamAV

Kevin van Zonneveld

Cofundador · Amsterdã, Países Baixos · Mostrar biografia

Crie um monitor em Rust que faz a varredura de arquivos concluídos com um daemon ClamAV local. Ele informa resultados limpos, infectados e de erro do scanner sem mover nem excluir arquivos. Somente uma resposta explicitamente limpa conta como limpa.

Por que monitoramento em tempo real e Rust?

Eventos do sistema de arquivos podem chegar antes que o processo que grava o arquivo tenha terminado. Um atraso ou um tamanho de arquivo inalterado não prova a conclusão. Por isso, este exemplo usa um contrato com o produtor: grave um arquivo com nome único terminado em .part, feche-o e depois renomeie-o atomicamente para .ready no mesmo diretório. Nunca modifique um arquivo .ready. Mantenha o diretório privado para a aplicação e o produtor confiável dela.

O monitor faz a varredura apenas de arquivos .ready no nível superior do diretório. Ele é uma ferramenta de notificação, não uma barreira de controle de acesso: um resultado descreve os bytes verificados e não autoriza uma leitura posterior de um caminho mutável.

Pré-requisitos

Este passo a passo usa Linux e Bash, com Rust e Cargo 1.98.1 e ClamAV 1.5.4. Tenha um daemon clamd local com um banco de assinaturas atualizado e um socket Unix acessível ao processo Rust. Os comandos abaixo são para Linux; o cliente usa sockets Unix.

Configuração do daemon do ClamAV (clamd)

No Debian ou Ubuntu, o guia de pacotes do ClamAV⁠ lista os pacotes do daemon e do atualizador de definições:

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

No clamd.conf da distribuição, mantenha as configurações de banco de dados e de usuário de serviço e configure um socket local. Este é um fragmento de configuração; garanta que nenhuma diretiva TCPSocket continue ativada:

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

Conceda ao usuário de serviço da aplicação acesso ao grupo do socket e depois reinicie o serviço. Deixe que o serviço FreshClam da distribuição mantenha o banco de dados atualizado. Não execute um segundo atualizador manual contra o banco de dados bloqueado por ele.

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

A documentação do protocolo do ClamAV⁠ descreve o socket e o enquadramento do INSTREAM. O Clamd não tem autenticação por TCP⁠; este exemplo usa apenas um socket local controlado por permissões. Configure os limites de contêineres de arquivos e de varredura conforme a sua carga de trabalho. Com AlertExceedsMax ativado, os alertas de limite excedido do daemon⁠ não devem ser tratados como limpos.

Configuração do projeto

Comece em um diretório gravável fora de um projeto Cargo existente. O Cargo pode adicionar um novo pacote a um workspace ancestral⁠, e a configuração é herdada dos diretórios pai⁠. Este bloco Bash recusa esses locais antes de criar qualquer coisa. Ele só entra no novo projeto depois que cargo new for bem-sucedido; um diretório realtime_virus_scanner existente não é alterado.

(
    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

Depois de uma configuração bem-sucedida, substitua o Cargo.toml do novo projeto por:

[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"] }

Mantenha o Cargo.lock na sua aplicação. O exemplo implementa diretamente com o Tokio a pequena troca INSTREAM, em vez de depender de um wrapper de cliente ClamAV separado.

Interface de linha de comando com clap

A CLI aceita --directory e --socket. Adicione --once para fazer a varredura dos arquivos concluídos existentes e encerrar: o código de saída zero significa que todos os arquivos selecionados estavam limpos, e o código um significa que ocorreu uma infecção ou um erro do scanner. Uma seleção vazia também encerra com código zero; isso não diz nada sobre arquivos .part inacabados. O modo de observação repete as varreduras até Ctrl+C e informa cada resultado.

Monitoramento de diretórios em tempo real com notify

Um watcher recomendado⁠ desperta o scanner após um evento de criação, modificação ou remoção. Um canal de uma única posição agrupa os eventos, pois cada despertar refaz a varredura do diretório. Um intervalo de reconciliação de cinco segundos também encontra arquivos quando eventos são perdidos. Eventos de leitura/acesso são ignorados para evitar que as varreduras disparem a si mesmas.

Varredura assíncrona de vírus com Tokio

Cada varredura obtém um snapshot limitado, transmite-o em blocos de 64 KiB e lê no máximo 4097 bytes de resposta. Somente a resposta limpa exata, única e terminada é aceita. Respostas vazias, truncadas, grandes demais, múltiplas ou não reconhecidas são tratadas como falha (fail closed). Uma infecção ou uma detecção de limite do engine não é considerada limpa. Um prazo de dez segundos cobre a leitura do arquivo e a comunicação com o clamd.

Juntando tudo: src/main.rs

Salve 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);
        }
    }
}

No diretório do projeto, compile e inicie o monitor com uma nova caixa de entrada privada. Cada etapa precisa ser bem-sucedida antes que a próxima seja executada, então um build com falha não pode iniciar um executável antigo:

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

Altere o argumento do socket se o seu daemon usar outro caminho. Deixe este terminal em execução. Em outro terminal, no diretório do projeto, publique um arquivo concluído usando um par de nomes ainda não utilizado:

(
    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
)

Uma varredura limpa imprime Clean Some("example.ready"), possivelmente mais de uma vez, porque as reconciliações refazem a varredura dos arquivos concluídos. O bloco produtor retorna um código diferente de zero se algum dos nomes já existir, sem substituir o conteúdo. Use nomes novos para os arquivos seguintes e não execute produtores simultaneamente com os mesmos nomes. O comando printf fecha o arquivo antes que mv o publique.

Pressione Ctrl+C no terminal do monitor para interrompê-lo. Para fazer uma única varredura da mesma caixa de entrada, execute:

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

Tratamento de erros e notificações

ScannerError significa que a varredura falhou, inclusive por socket indisponível, entrada acima de 10 MiB, symlink ou prazo expirado. Infected também abrange os alertas de limite excedido configurados no ClamAV. Nenhum dos dois resultados deve liberar um arquivo para consumidores downstream. O exemplo deixa deliberadamente todos os arquivos no lugar, para que uma falha de varredura não possa excluir um upload.

Ctrl+C descarta a future da varredura ativa, fecha o socket dela, libera o watcher e encerra com código diferente de zero. No modo de observação, uma varredura com falha é tentada novamente em uma reconciliação posterior. Use --once quando quem chama precisar de um código de saída de falha para um lote não limpo.

Considerações de desempenho

Há uma varredura ativa por vez, um despertar pendente, um snapshot limitado por arquivo e uma resposta limitada. A enumeração do diretório não acumula uma fila de caminhos. Refazer a varredura de todos os arquivos concluídos sacrifica throughput em troca de um exemplo pequeno e recuperável; um serviço maior deve registrar o trabalho de forma durável e associar cada veredito ao objeto imutável exato que foi verificado.

Este monitor não é recursivo, e o intervalo de cinco segundos é um cronograma de reconciliação, não um prazo garantido de detecção. Lotes grandes e varreduras lentas levam mais tempo. Mantenha o banco de assinaturas atualizado e teste os limites de contêineres de arquivos do seu daemon como parte da operação do serviço.

Conclusão

Somente produtores confiáveis devem publicar arquivos .ready, e os consumidores não devem tratar este monitor como permissão para servi-los. Clean significa que o engine do ClamAV configurado não encontrou nenhuma detecção nos bytes verificados, e não que o arquivo seja inofensivo. Mantenha as assinaturas atualizadas e escolha limites do daemon adequados aos formatos que você aceita.

A Transloadit também usa o ClamAV para filtrar arquivos por meio do seu Robot 🤖 /file/virusscan.