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
This walkthrough uses Linux and Bash, with Rust and Cargo 1.98.1 and ClamAV 1.5.4. Have a local
clamd daemon with an updated signature database and a Unix socket accessible to the Rust process.
The commands below target Linux; the client uses Unix sockets.
Setting up ClamAV daemon (clamd)
On Debian or Ubuntu, the ClamAV package guide lists the daemon and definition updater packages:
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. With AlertExceedsMax enabled, the daemon’s limit-exceeded alerts
must not be treated as clean.
Project setup
Start in a writable directory outside an existing Cargo project. Cargo can add a new package to an
ancestor workspace, and
configuration is inherited from parent directories.
This Bash block refuses those locations before creating anything. It enters the new project only
after cargo new succeeds; an existing realtime_virus_scanner directory is left alone.
(
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
After successful setup, replace the new project’s 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.
An empty selection also exits zero; it says nothing about unfinished .part files. 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);
}
}
}
From the project directory, build and start the monitor with a new private inbox. Each step must succeed before the next runs, so a failed build cannot start an old executable:
mkdir -m 700 inbox &&
cargo build --jobs 2 --target-dir target &&
./target/debug/realtime_virus_scanner --directory inbox --socket /var/run/clamav/clamd.ctl
Change the socket argument if your daemon uses a different path. Leave this terminal running. In another terminal, from the project directory, publish a finished file using an unused pair of names:
(
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
)
A clean scan prints Clean Some("example.ready"), possibly more than once because reconciliations
rescan completed files. The producer block returns nonzero if either name already exists, without
replacing its contents. Use fresh names for later files, and do not run producers concurrently with
the same names. The printf command closes the file before mv publishes it.
Press Ctrl+C in the monitor terminal to stop it. To scan the same inbox once, run:
./target/debug/realtime_virus_scanner --directory inbox --socket /var/run/clamav/clamd.ctl --once
Error handling and Notifications
ScannerError means the scan failed, including an unavailable socket, an input over 10 MiB, a
symlink, or a deadline expiry. Infected also covers ClamAV’s configured limit-exceeded alerts.
Neither result should release a file to downstream consumers. 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
Only trusted producers should publish .ready files, and consumers must not treat this monitor as
permission to serve them. Clean means the configured ClamAV engine found no detection in the bytes
it scanned, not that the file is harmless. Keep signatures updated and choose daemon limits for the
formats you accept.
Transloadit also uses ClamAV for file filtering through its 🤖 /file/virusscan Robot.