Deep Dive Antivirus in 2026: Tested & Compared

// eBPF On-Access Scan Pre-Filter for Antivirus Engines
// SPDX-License-Identifier: GPL-2.0
#include 
#include 
#include 
#include 
#include 
#include 

// Define ring buffer for passing filtered file paths to userspace AV daemon
struct {
    __uint(type, BPF_MAP_TYPE_RINGBUF);
    __uint(max_entries, 1 << 24); // 16MB ring buffer
} filtered_files SEC(".maps");

// Allowlist of file extensions to skip scanning (configurable via map)
struct allowed_ext {
    char ext[8];
    __u8 len;
};

struct {
    __uint(type, BPF_MAP_TYPE_HASH);
    __type(key, struct allowed_ext);
    __type(value, __u8);
    __uint(max_entries, 32);
} ext_allowlist SEC(".maps");

// Hook into vfs_open syscall to intercept file opens before AV engine
SEC("kprobe/vfs_open")
int BPF_KPROBE(vfs_open_pre_filter, struct path *path, struct file *file)
{
    struct allowed_ext ext_key = {0};
    char filename[256];
    __u8 *allowed;
    int ret;

    // Read full filename from kernel path struct
    ret = bpf_d_path(&path->mnt, &path->dentry, filename, sizeof(filename));
    if (ret < 0) {
        bpf_printk("vfs_open_pre_filter: failed to read path: %d", ret);
        return 0; // Pass to AV engine on error to avoid bypass
    }

    // Extract file extension (last 7 chars after .)
    char *dot = bpf_strrchr(filename, '.');
    if (!dot) {
        // No extension, pass to AV
        goto submit;
    }
    __u8 ext_len = bpf_strnlen(dot + 1, 7);
    if (ext_len > 7) {
        ext_len = 7;
    }
    bpf_memcpy(ext_key.ext, dot + 1, ext_len);
    ext_key.len = ext_len;

    // Check if extension is allowlisted
    allowed = bpf_map_lookup_elem(&ext_allowlist, &ext_key);
    if (allowed && *allowed == 1) {
        bpf_printk("vfs_open_pre_filter: skipping allowed ext: %s", ext_key.ext);
        return 0; // Skip AV scan, file is safe
    }

submit:
    // Submit file path to userspace AV daemon via ring buffer
    char *buf = bpf_ringbuf_reserve(&filtered_files, sizeof(filename), 0);
    if (!buf) {
        bpf_printk("vfs_open_pre_filter: ringbuf full, passing to AV");
        return 0;
    }
    bpf_memcpy(buf, filename, sizeof(filename));
    bpf_ringbuf_submit(buf, 0);
    return 0; // Still pass to AV engine for full scan
}

char _license[] SEC("license") = "GPL";

// Rust-Based Antivirus Signature Parser (Memory-Safe)
// Implements ClamAV 2026 signature format v4 parsing
// Uses zero-copy deserialization to avoid heap allocations during scans
use std::error::Error;
use std::fs::File;
use std::io::{BufReader, Read, Seek, SeekFrom};
use std::path::Path;

// Signature header magic bytes: 0x43 0x4C 0x44 0x34 (CLD4)
const SIGNATURE_MAGIC: [u8; 4] = [0x43, 0x4C, 0x44, 0x34];
const MAX_SIGNATURE_SIZE: usize = 1024 * 1024; // 1MB max per signature

#[derive(Debug)]
pub enum ParseError {
    InvalidMagic,
    TruncatedData,
    UnsupportedVersion(u8),
    ChecksumMismatch(u32, u32),
}

#[derive(Debug, PartialEq)]
pub struct SignatureEntry {
    pub id: u64,
    pub threat_name: String,
    pub offset: u32,
    pub pattern: Vec,
    pub checksum: u32,
}

pub struct SignatureParser {
    reader: BufReader,
    current_offset: u64,
}

impl SignatureParser {
    pub fn new(reader: R) -> Self {
        Self {
            reader: BufReader::new(reader),
            current_offset: 0,
        }
    }

    /// Parse all signatures from the input stream, returning a vector of valid entries
    pub fn parse_all(&mut self) -> Result, Box> {
        let mut entries = Vec::new();
        // Verify magic header first
        let mut magic_buf = [0u8; 4];
        self.reader.read_exact(&mut magic_buf)?;
        if magic_buf != SIGNATURE_MAGIC {
            return Err(Box::new(ParseError::InvalidMagic));
        }
        self.current_offset += 4;

        // Read version byte (must be 4 for 2026 format)
        let mut version_buf = [0u8; 1];
        self.reader.read_exact(&mut version_buf)?;
        let version = version_buf[0];
        if version != 4 {
            return Err(Box::new(ParseError::UnsupportedVersion(version)));
        }
        self.current_offset += 1;

        // Parse entries until EOF
        loop {
            match self.parse_next_entry() {
                Ok(Some(entry)) => entries.push(entry),
                Ok(None) => break, // EOF
                Err(e) => {
                    eprintln!("Warning: failed to parse entry at offset {}: {}", self.current_offset, e);
                    // Skip to next entry by seeking to next magic or EOF
                    self.skip_to_next_entry()?;
                }
            }
        }

        Ok(entries)
    }

    fn parse_next_entry(&mut self) -> Result, Box> {
        let start_offset = self.current_offset;
        // Read entry length (4 bytes little-endian)
        let mut len_buf = [0u8; 4];
        if self.reader.read(&mut len_buf)? != 4 {
            return Ok(None); // EOF
        }
        let entry_len = u32::from_le_bytes(len_buf) as usize;
        if entry_len > MAX_SIGNATURE_SIZE {
            return Err(Box::new(std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                format!("Entry size {} exceeds max {}", entry_len, MAX_SIGNATURE_SIZE)
            )));
        }
        self.current_offset += 4;

        // Read full entry data
        let mut entry_data = vec![0u8; entry_len];
        self.reader.read_exact(&mut entry_data)?;
        self.current_offset += entry_len as u64;

        // Parse entry fields: id (8), threat_name_len (2), threat_name, offset (4), pattern_len (4), pattern, checksum (4)
        let mut cursor = 0;
        if cursor + 8 > entry_data.len() { return Err(Box::new(ParseError::TruncatedData)); }
        let id = u64::from_le_bytes(entry_data[cursor..cursor+8].try_into()?);
        cursor += 8;

        if cursor + 2 > entry_data.len() { return Err(Box::new(ParseError::TruncatedData)); }
        let threat_name_len = u16::from_le_bytes(entry_data[cursor..cursor+2].try_into()?) as usize;
        cursor += 2;

        if cursor + threat_name_len > entry_data.len() { return Err(Box::new(ParseError::TruncatedData)); }
        let threat_name = String::from_utf8(entry_data[cursor..cursor+threat_name_len].to_vec())?;
        cursor += threat_name_len;

        if cursor + 4 > entry_data.len() { return Err(Box::new(ParseError::TruncatedData)); }
        let offset = u32::from_le_bytes(entry_data[cursor..cursor+4].try_into()?);
        cursor += 4;

        if cursor + 4 > entry_data.len() { return Err(Box::new(ParseError::TruncatedData)); }
        let pattern_len = u32::from_le_bytes(entry_data[cursor..cursor+4].try_into()?) as usize;
        cursor += 4;

        if cursor + pattern_len > entry_data.len() { return Err(Box::new(ParseError::TruncatedData)); }
        let pattern = entry_data[cursor..cursor+pattern_len].to_vec();
        cursor += pattern_len;

        if cursor + 4 > entry_data.len() { return Err(Box::new(ParseError::TruncatedData)); }
        let checksum = u32::from_le_bytes(entry_data[cursor..cursor+4].try_into()?);

        // Verify checksum (CRC32 of entry data before checksum field)
        let calculated_checksum = crc32fast::hash(&entry_data[..entry_data.len()-4]);
        if calculated_checksum != checksum {
            return Err(Box::new(ParseError::ChecksumMismatch(calculated_checksum, checksum)));
        }

        Ok(Some(SignatureEntry { id, threat_name, offset, pattern, checksum }))
    }

    fn skip_to_next_entry(&mut self) -> Result<(), Box> {
        // Seek forward until we find next possible entry start or EOF
        let mut buf = [0u8; 1024];
        loop {
            let bytes_read = self.reader.read(&mut buf)?;
            if bytes_read == 0 { break; }
            self.current_offset += bytes_read as u64;
        }
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::io::Cursor;

    #[test]
    fn test_parse_valid_signature() {
        let test_data = [
            SIGNATURE_MAGIC.as_ref(),
            &[4], // version 4
            &[0x00, 0x00, 0x00, 0x20], // entry len 32
            // entry data: id=1, threat_name_len=5, "EICAR", offset=0, pattern_len=4, [0xDE,0xAD,0xBE,0xEF], checksum
        ].concat();
        let cursor = Cursor::new(test_data);
        let mut parser = SignatureParser::new(cursor);
        let entries = parser.parse_all().unwrap();
        assert_eq!(entries.len(), 1);
        assert_eq!(entries[0].threat_name, "EICAR");
    }
}

// Antivirus Scan Latency Benchmark Tool (Python 3.12+)
// Measures on-access scan latency, CPU usage, and false positive rates
// Supports Linux (fanotify), Windows (FSFilter), macOS (FSEvents) via cross-platform wrappers
import os
import sys
import time
import json
import psutil
import argparse
import subprocess
from pathlib import Path
from typing import Dict, List, Optional
from dataclasses import dataclass, field
from datetime import datetime

@dataclass
class BenchmarkConfig:
    test_dir: Path
    file_sizes: List[int] = field(default_factory=lambda: [1024, 1024*1024, 10*1024*1024]) # 1KB, 1MB, 10MB
    file_count: int = 100
    iterations: int = 3
    av_daemon_pid: Optional[int] = None
    output_file: Path = Path("av_benchmark_results.json")

@dataclass
class BenchmarkResult:
    timestamp: str
    tool_name: str
    tool_version: str
    scan_latency_ms: Dict[str, float] # p50, p95, p99
    cpu_usage_percent: float
    false_positives: int
    files_scanned: int

class AVBenchmarker:
    def __init__(self, config: BenchmarkConfig):
        self.config = config
        self.results: List[BenchmarkResult] = []
        self._validate_config()

    def _validate_config(self) -> None:
        if not self.config.test_dir.exists():
            raise FileNotFoundError(f"Test directory {self.config.test_dir} does not exist")
        if os.geteuid() != 0:
            print("Warning: Running without root privileges may limit on-access scan measurement accuracy")

    def _get_av_metadata(self) -> Dict[str, str]:
        # Try to detect installed AV tools and their versions
        av_tools = {
            "clamav": ["clamdscan", "--version"],
            "windows-defender": ["powershell", "-Command", "Get-MpComputerStatus | Select-Object -ExpandProperty AMProductVersion"],
            "sophos": ["savdversion"],
        }
        metadata = {}
        for tool, cmd in av_tools.items():
            try:
                result = subprocess.run(cmd, capture_output=True, text=True, timeout=5)
                if result.returncode == 0:
                    metadata[tool] = result.stdout.strip()
            except (subprocess.TimeoutExpired, FileNotFoundError):
                continue
        return metadata

    def _generate_test_files(self) -> List[Path]:
        # Create test files of varying sizes with random content
        test_files = []
        self.config.test_dir.mkdir(parents=True, exist_ok=True)
        for size in self.config.file_sizes:
            for i in range(self.config.file_count // len(self.config.file_sizes)):
                file_path = self.config.test_dir / f"test_{size}_{i}.bin"
                if not file_path.exists():
                    with open(file_path, "wb") as f:
                        f.write(os.urandom(size))
                test_files.append(file_path)
        return test_files

    def _measure_scan_latency(self, test_files: List[Path]) -> Dict[str, float]:
        # Measure time from file open to AV scan completion using fanotify hooks
        latencies = []
        for file_path in test_files:
            start = time.perf_counter()
            # Trigger on-access scan by opening file in read mode
            try:
                with open(file_path, "rb") as f:
                    f.read(1) # Read 1 byte to trigger scan
            except IOError as e:
                print(f"Failed to open {file_path}: {e}")
                continue
            end = time.perf_counter()
            latencies.append((end - start) * 1000) # Convert to ms

        if not latencies:
            return {"p50": 0.0, "p95": 0.0, "p99": 0.0}
        latencies.sort()
        p50 = latencies[len(latencies)//2]
        p95 = latencies[int(len(latencies)*0.95)]
        p99 = latencies[int(len(latencies)*0.99)]
        return {"p50": p50, "p95": p95, "p99": p99}

    def _measure_cpu_usage(self, duration_sec: int = 10) -> float:
        # Measure AV daemon CPU usage over specified duration
        if not self.config.av_daemon_pid:
            return 0.0
        try:
            process = psutil.Process(self.config.av_daemon_pid)
            cpu_percentages = []
            for _ in range(duration_sec):
                cpu_percentages.append(process.cpu_percent(interval=1))
            return sum(cpu_percentages) / len(cpu_percentages)
        except psutil.NoSuchProcess:
            print(f"AV daemon with PID {self.config.av_daemon_pid} not found")
            return 0.0

    def run_benchmark(self) -> None:
        av_metadata = self._get_av_metadata()
        if not av_metadata:
            raise RuntimeError("No supported antivirus tools detected")
        test_files = self._generate_test_files()
        print(f"Generated {len(test_files)} test files, starting benchmark...")

        for tool_name, tool_version in av_metadata.items():
            print(f"Benchmarking {tool_name} version {tool_version}...")
            for _ in range(self.config.iterations):
                scan_latency = self._measure_scan_latency(test_files)
                cpu_usage = self._measure_cpu_usage()
                # TODO: Implement false positive measurement using EICAR test files
                false_positives = 0
                result = BenchmarkResult(
                    timestamp=datetime.utcnow().isoformat(),
                    tool_name=tool_name,
                    tool_version=tool_version,
                    scan_latency_ms=scan_latency,
                    cpu_usage_percent=cpu_usage,
                    false_positives=false_positives,
                    files_scanned=len(test_files)
                )
                self.results.append(result)

        self._save_results()

    def _save_results(self) -> None:
        with open(self.config.output_file, "w") as f:
            json.dump([vars(r) for r in self.results], f, indent=2)
        print(f"Results saved to {self.config.output_file}")

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Antivirus Scan Latency Benchmark Tool")
    parser.add_argument("--test-dir", type=Path, default=Path("/tmp/av_benchmark"))
    parser.add_argument("--file-count", type=int, default=100)
    parser.add_argument("--iterations", type=int, default=3)
    parser.add_argument("--av-pid", type=int, help="PID of AV daemon to monitor CPU usage")
    parser.add_argument("--output", type=Path, default=Path("av_benchmark_results.json"))
    args = parser.parse_args()

    config = BenchmarkConfig(
        test_dir=args.test_dir,
        file_count=args.file_count,
        iterations=args.iterations,
        av_daemon_pid=args.av_pid,
        output_file=args.output
    )

    try:
        benchmarker = AVBenchmarker(config)
        benchmarker.run_benchmark()
    except Exception as e:
        print(f"Benchmark failed: {e}", file=sys.stderr)
        sys.exit(1)

// Userspace snippet to update eBPF allowlist at runtime
#include 
struct allowed_ext ext = {.ext = "jar", .len = 3};
__u8 val = 1;
bpf_map_update_elem(map_fd, &ext, &val, BPF_ANY);

// Fuzz target for Rust signature parser
#![no_main]
use libfuzzer_sys::fuzz_target;
fuzz_target!(|data: &[u8]| {
    let cursor = std::io::Cursor::new(data);
    let mut parser = SignatureParser::new(cursor);
    let _ = parser.parse_all();
});

// Run benchmark for CI/CD workload
python3 av_benchmark.py --test-dir /var/cache/ci-artifacts --file-count 500 --av-pid $(pgrep clamd)