teamcraft/
malware.rs

1//! Optional malware scanning of uploads with a ClamAV daemon (`clamd`),
2//! spoken to over its `INSTREAM` protocol on TCP (`host:port`) or, on Unix,
3//! a socket (`unix:/run/clamav/clamd.ctl`). No ClamAV code is linked.
4//!
5//! The vault scans every file before it moves it to its content address
6//! (`Vault::with_scanner`), so infected content never reaches `objects/`.
7//! The check fails closed: when the daemon cannot be reached or answers
8//! with an error, the upload is refused (503) instead of stored unscanned.
9//! Files above the configured size (`TEAMCRAFT_CLAMD_MAX_MB`, clamd's own
10//! `StreamMaxLength`) are stored without a scan and logged.
11
12use std::{path::Path, time::Duration};
13
14use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
15
16use crate::error::{AppError, AppResult};
17
18/// Bytes sent per `INSTREAM` chunk.
19const CHUNK: usize = 256 * 1024;
20/// Longest reply read from the daemon.
21const MAX_REPLY: usize = 4096;
22
23/// Connection settings of the daemon (`TEAMCRAFT_CLAMD_*`).
24#[derive(Clone, Debug, PartialEq, Eq)]
25pub struct ScannerConfig {
26    /// `host:port` or `unix:/path/to/socket`.
27    pub addr: String,
28    /// Larger files are not scanned (clamd refuses streams above its
29    /// `StreamMaxLength`, 25 MiB by default).
30    pub max_bytes: u64,
31    /// Time for one whole scan (connect, send, answer).
32    pub timeout: Duration,
33}
34
35/// What the daemon said about a file.
36#[derive(Clone, Debug, PartialEq, Eq)]
37pub enum Verdict {
38    Clean,
39    /// The signature name, e.g. `Win.Test.EICAR_HDB-1`.
40    Infected(String),
41    /// Larger than [`ScannerConfig::max_bytes`]: stored without a scan.
42    TooLarge,
43}
44
45/// Reads the `zINSTREAM` answer: `stream: OK`, `stream: <name> FOUND` or
46/// `… ERROR`.
47pub fn parse_reply(raw: &[u8]) -> Result<Verdict, String> {
48    let text = String::from_utf8_lossy(raw);
49    let text = text.trim_end_matches(['\0', '\n', '\r', ' ']).trim();
50    let body = text.strip_prefix("stream:").map(str::trim).unwrap_or(text);
51    if body == "OK" {
52        return Ok(Verdict::Clean);
53    }
54    if let Some(name) = body.strip_suffix("FOUND") {
55        let name = name.trim();
56        return Ok(Verdict::Infected(if name.is_empty() {
57            "unknown".into()
58        } else {
59            name.chars().take(200).collect()
60        }));
61    }
62    Err(if body.is_empty() {
63        "empty answer".into()
64    } else {
65        body.chars().take(200).collect()
66    })
67}
68
69/// Sends a file as one `INSTREAM` and reads the answer.
70async fn instream<S: AsyncRead + AsyncWrite + Unpin>(
71    stream: &mut S,
72    path: &Path,
73) -> std::io::Result<Vec<u8>> {
74    let mut file = tokio::fs::File::open(path).await?;
75    stream.write_all(b"zINSTREAM\0").await?;
76    let mut buf = vec![0u8; CHUNK];
77    loop {
78        let n = file.read(&mut buf).await?;
79        if n == 0 {
80            break;
81        }
82        stream.write_all(&(n as u32).to_be_bytes()).await?;
83        stream.write_all(&buf[..n]).await?;
84    }
85    stream.write_all(&0u32.to_be_bytes()).await?;
86    stream.flush().await?;
87    let mut reply = Vec::new();
88    let mut byte = [0u8; 512];
89    loop {
90        let n = stream.read(&mut byte).await?;
91        if n == 0 {
92            break;
93        }
94        reply.extend_from_slice(&byte[..n]);
95        if reply.contains(&0) || reply.len() >= MAX_REPLY {
96            break;
97        }
98    }
99    Ok(reply)
100}
101
102async fn scan_raw(config: &ScannerConfig, path: &Path) -> std::io::Result<Vec<u8>> {
103    if let Some(socket) = config.addr.strip_prefix("unix:") {
104        #[cfg(unix)]
105        {
106            let mut stream = tokio::net::UnixStream::connect(socket).await?;
107            return instream(&mut stream, path).await;
108        }
109        #[cfg(not(unix))]
110        {
111            let _ = socket;
112            return Err(std::io::Error::new(
113                std::io::ErrorKind::Unsupported,
114                "unix sockets are not available on this platform",
115            ));
116        }
117    }
118    let mut stream = tokio::net::TcpStream::connect(&config.addr).await?;
119    instream(&mut stream, path).await
120}
121
122/// Scans a file. Errors (daemon unreachable, timeout, an `ERROR` answer)
123/// are returned as `Unavailable`: the caller refuses the upload.
124pub async fn scan_file(config: &ScannerConfig, path: &Path, size: u64) -> AppResult<Verdict> {
125    if size > config.max_bytes {
126        tracing::info!(
127            size,
128            max = config.max_bytes,
129            "file larger than TEAMCRAFT_CLAMD_MAX_MB stored without a malware scan"
130        );
131        return Ok(Verdict::TooLarge);
132    }
133    let unavailable = |detail: String| {
134        tracing::warn!(addr = %config.addr, "malware scan failed: {detail}");
135        AppError::Unavailable(
136            "The malware scanner is not available; the file was not stored. Try again later."
137                .into(),
138        )
139    };
140    let raw = tokio::time::timeout(config.timeout, scan_raw(config, path))
141        .await
142        .map_err(|_| unavailable("timed out".into()))?
143        .map_err(|e| unavailable(e.to_string()))?;
144    parse_reply(&raw).map_err(unavailable)
145}
146
147/// Scans a file about to enter the vault and refuses infected content (400).
148pub async fn ensure_clean(config: &ScannerConfig, path: &Path, size: u64) -> AppResult<()> {
149    match scan_file(config, path, size).await? {
150        Verdict::Clean | Verdict::TooLarge => Ok(()),
151        Verdict::Infected(name) => {
152            tracing::warn!(signature = %name, "upload refused: malware found");
153            Err(AppError::BadRequest(format!(
154                "The file was refused: the malware scanner found {name}."
155            )))
156        }
157    }
158}
159
160#[cfg(test)]
161pub(crate) mod tests {
162    use super::*;
163    use tokio::net::TcpListener;
164
165    #[test]
166    fn replies() {
167        assert_eq!(parse_reply(b"stream: OK\0"), Ok(Verdict::Clean));
168        assert_eq!(parse_reply(b"OK\n"), Ok(Verdict::Clean));
169        assert_eq!(
170            parse_reply(b"stream: Win.Test.EICAR_HDB-1 FOUND\0"),
171            Ok(Verdict::Infected("Win.Test.EICAR_HDB-1".into()))
172        );
173        assert!(parse_reply(b"INSTREAM size limit exceeded. ERROR\0").is_err());
174        assert!(parse_reply(b"").is_err());
175        assert_eq!(
176            parse_reply(b"stream:  FOUND\0"),
177            Ok(Verdict::Infected("unknown".into()))
178        );
179    }
180
181    /// A fake clamd: reads one INSTREAM and answers FOUND when the content
182    /// holds "EICAR", OK otherwise.
183    pub(crate) async fn fake_clamd() -> String {
184        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
185        let addr = listener.local_addr().unwrap().to_string();
186        tokio::spawn(async move {
187            while let Ok((mut socket, _)) = listener.accept().await {
188                tokio::spawn(async move {
189                    let mut cmd = [0u8; 10];
190                    socket.read_exact(&mut cmd).await.unwrap();
191                    assert_eq!(&cmd, b"zINSTREAM\0");
192                    let mut content = Vec::new();
193                    loop {
194                        let mut len = [0u8; 4];
195                        socket.read_exact(&mut len).await.unwrap();
196                        let len = u32::from_be_bytes(len) as usize;
197                        if len == 0 {
198                            break;
199                        }
200                        let mut chunk = vec![0u8; len];
201                        socket.read_exact(&mut chunk).await.unwrap();
202                        content.extend_from_slice(&chunk);
203                    }
204                    let found = content.windows(5).any(|w| w == b"EICAR");
205                    let reply: &[u8] = if found {
206                        b"stream: Eicar-Signature FOUND\0"
207                    } else {
208                        b"stream: OK\0"
209                    };
210                    socket.write_all(reply).await.unwrap();
211                });
212            }
213        });
214        addr
215    }
216
217    fn temp_dir() -> std::path::PathBuf {
218        let dir = std::env::temp_dir().join(format!("teamcraft-malware-{}", uuid::Uuid::new_v4()));
219        std::fs::create_dir_all(&dir).unwrap();
220        dir
221    }
222
223    fn config(addr: String) -> ScannerConfig {
224        ScannerConfig {
225            addr,
226            max_bytes: 1024 * 1024,
227            timeout: Duration::from_secs(5),
228        }
229    }
230
231    #[tokio::test]
232    async fn scans_through_the_instream_protocol() {
233        let cfg = config(fake_clamd().await);
234        let dir = temp_dir();
235        let clean = dir.join("clean.step");
236        std::fs::write(&clean, vec![b'x'; 600_000]).unwrap(); // several chunks
237        let bad = dir.join("bad.step");
238        std::fs::write(&bad, b"X5O!P%@AP EICAR test").unwrap();
239
240        assert_eq!(
241            scan_file(&cfg, &clean, 600_000).await.unwrap(),
242            Verdict::Clean
243        );
244        assert_eq!(
245            scan_file(&cfg, &bad, 20).await.unwrap(),
246            Verdict::Infected("Eicar-Signature".into())
247        );
248        assert!(ensure_clean(&cfg, &clean, 600_000).await.is_ok());
249        match ensure_clean(&cfg, &bad, 20).await {
250            Err(AppError::BadRequest(m)) => assert!(m.contains("Eicar-Signature"), "{m}"),
251            other => panic!("{other:?}"),
252        }
253        // above the size limit: not sent to the daemon
254        assert_eq!(
255            scan_file(&cfg, &bad, 2 * 1024 * 1024).await.unwrap(),
256            Verdict::TooLarge
257        );
258        let _ = std::fs::remove_dir_all(dir);
259    }
260
261    #[tokio::test]
262    async fn an_unreachable_scanner_fails_closed() {
263        // a port nothing listens on
264        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
265        let addr = listener.local_addr().unwrap().to_string();
266        drop(listener);
267        let dir = temp_dir();
268        let file = dir.join("a.step");
269        std::fs::write(&file, b"content").unwrap();
270        assert!(matches!(
271            ensure_clean(&config(addr), &file, 7).await,
272            Err(AppError::Unavailable(_))
273        ));
274        let _ = std::fs::remove_dir_all(dir);
275    }
276}
teamcraft/malware.rs

teamcraft/
malware.rs
