Real-time directory monitoring and virus scanning with Rust and ClamAV
Build a Rust monitor that scans completed files with a local ClamAV daemon. It reports clean, infected, and scanner-error outcomes without moving or deleting files. Only an explicit clean response counts as clean.
Why real-time monitoring and Rust?
Filesystem events can arrive before a writer has finished. A delay or an unchanged file size does not prove completion. This example therefore uses a producer contract: write a unique file ending in .part, close it, then atomically rename it to .ready in the same directory. Never modify a .ready file. Keep the directory private to the application and its trusted producer.
The monitor scans only top-level .ready files. It is a notification tool, not an access-control gate: a result describes the bytes scanned and does not authorize a later read of a mutable path.
Prerequisites
Use Linux or macOS with Rust and Cargo. The program was compiled with Rust 1.98.1 and tested against ClamAV 1.5.4. A local clamd daemon must have a current signature database and a Unix socket accessible to the Rust process.
Setting up ClamAV daemon (clamd)
On Debian or Ubuntu, install the daemon and definition updater:
sudo apt-get update
sudo apt-get install clamav-daemon clamav-freshclam
In the distribution’s clamd.conf, retain its database and service-user settings and configure a local socket. This is a configuration fragment; ensure no TCPSocket directive remains enabled:
LocalSocket /var/run/clamav/clamd.ctl
LocalSocketMode 660
StreamMaxLength 10M
MaxFileSize 10M
MaxScanSize 20M
AlertExceedsMax yes
Grant the application’s service user access to the socket’s group, then restart the service. Let the distribution’s FreshClam service keep the database updated. Do not run a second manual updater against its locked database.
sudo systemctl enable --now clamav-freshclam
sudo systemctl restart clamav-daemon
The ClamAV protocol documentation describes the socket and INSTREAM framing. Clamd has no TCP authentication; this example uses only a permission-controlled local socket. Configure archive and scan limits for your workload; limit-exceeded detections must not be treated as clean.
Project setup
cargo new realtime_virus_scanner --edition 2021 --vcs none
cd realtime_virus_scanner
Replace Cargo.toml with:
[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"] }
Retain Cargo.lock in your application. The example implements the small INSTREAM exchange directly with Tokio, rather than depending on a separate ClamAV client wrapper.
Command-line interface with clap
The CLI accepts --directory and --socket. Add --once to scan the existing completed files and exit: zero means every selected file was clean, one means an infection or scanner error occurred. Watch mode repeats scans until Ctrl-C and reports each result.
Real-time directory monitoring with notify
A recommended watcher wakes the scanner after a create, modify, or remove event. A one-slot channel coalesces events because each wake rescans the directory. A five-second reconciliation interval also finds files when events are missed. Read/access events are ignored to avoid scans triggering themselves.
Asynchronous virus scanning with Tokio
Each scan takes a bounded snapshot, streams it in 64 KiB chunks, and reads at most 4097 response bytes. Only the exact, single, terminated clean reply is accepted. Empty, truncated, oversized, multiple, or unrecognized replies fail closed. An infection or engine-limit detection is non-clean. A ten-second deadline covers reading the file and communicating with clamd.
Putting it all together: src/main.rs
Save this complete program as 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);
}
}
}
Build and start the monitor with an empty private inbox:
mkdir -m 700 inbox
cargo build
cargo run -- --directory inbox --socket /var/run/clamav/clamd.ctl
In another terminal, from the project directory, publish a finished file with a fresh name:
printf 'A completed example file.\n' > inbox/example.part
mv inbox/example.part inbox/example.ready
A clean scan prints Clean. Keep .part files out of the scanner’s input set; only rename after the writer has closed the file, and do not reuse or alter a published .ready path.
Error handling and Notifications
ScannerError means the file has not been cleared. Infected also covers ClamAV’s configured limit-exceeded alerts. Neither result should trigger file publication. The example deliberately leaves every file in place so a scan failure cannot delete an upload.
Ctrl-C drops the active scan future, closes its socket, releases the watcher, and exits nonzero. In watch mode, a failed scan is retried during a later reconciliation. Use --once when a caller needs a failing exit code for a non-clean batch.
Performance considerations
There is one active scan at a time, one pending wake, a bounded per-file snapshot, and a bounded reply. Directory enumeration does not accumulate a queue of paths. Scanning all completed files again trades throughput for a small and recoverable example; a larger service should track work durably and associate each verdict with the exact immutable object it scanned.
This monitor is not recursive, and a five-second interval is a reconciliation schedule rather than a guaranteed detection deadline. Large batches and slow scans take longer. Maintain signatures and test your daemon’s archive limits as part of operating the service.
Conclusion
The producer handoff prevents partial writes from being mistaken for complete files. Notify provides prompt wakeups, periodic rescans recover missed events, and explicit protocol validation keeps ambiguous scanner outcomes non-clean.
Transloadit also uses ClamAV for file filtering through its 🤖 /file/virusscan Robot.