Skip to main content

max / makenotwork

16.3 KB · 413 lines History Blame Raw
1 //! Layer 5: ClamAV daemon scanning via Unix socket.
2 //!
3 //! Connects to `clamd` using the INSTREAM protocol: sends file data in
4 //! length-prefixed chunks, reads the verdict. Optional, only runs if
5 //! CLAMAV_SOCKET env var is set.
6
7 use std::time::Duration;
8
9 use tokio::io::{AsyncReadExt, AsyncWriteExt};
10 use tokio::net::UnixStream;
11
12 use crate::constants;
13
14 use super::{ErrorPolicy, LayerResult, LayerVerdict};
15
16 /// External local-service layer. An `Error` here means `clamd` is down,
17 /// unreachable, or out of memory, operational problems that must not block
18 /// every upload across the platform. Fail open; PoM surfaces degraded health.
19 pub const ERROR_POLICY: ErrorPolicy = ErrorPolicy::FailOpen;
20
21 /// Verify the `clamd` socket is reachable and responsive. Used at startup
22 /// to refuse to boot if scanning is configured but the daemon is dead,
23 /// otherwise the FailOpen policy would silently pass every upload as Clean
24 /// for the lifetime of the process.
25 pub async fn ping(socket_path: &str) -> Result<(), String> {
26 let timeout = Duration::from_secs(2);
27 let work = async {
28 let mut stream = UnixStream::connect(socket_path)
29 .await
30 .map_err(|e| format!("connect: {e}"))?;
31 stream
32 .write_all(b"zPING\0")
33 .await
34 .map_err(|e| format!("write PING: {e}"))?;
35 let mut response = Vec::with_capacity(8);
36 stream
37 .take(16)
38 .read_to_end(&mut response)
39 .await
40 .map_err(|e| format!("read PONG: {e}"))?;
41 let reply = String::from_utf8_lossy(&response);
42 let reply = reply.trim_end_matches('\0').trim();
43 if reply == "PONG" {
44 Ok(())
45 } else {
46 Err(format!("unexpected PING response: {reply:?}"))
47 }
48 };
49 tokio::time::timeout(timeout, work)
50 .await
51 .map_err(|_| "ClamAV PING timed out after 2s".to_string())?
52 }
53
54 /// Maximum chunk size for INSTREAM protocol (ClamAV default)
55 const CHUNK_SIZE: usize = 8192;
56
57 /// Scan file data using a ClamAV daemon via Unix socket.
58 pub async fn scan_with_clamav(socket_path: &str, data: &[u8]) -> LayerResult {
59 let timeout = Duration::from_secs(constants::SCAN_CLAMAV_TIMEOUT_SECS);
60
61 match tokio::time::timeout(timeout, scan_instream(socket_path, data)).await {
62 Ok(Ok(result)) => result,
63 Ok(Err(e)) => LayerResult {
64 layer: "clamav",
65 verdict: LayerVerdict::Error,
66 detail: Some(format!("ClamAV error: {e}")),
67 },
68 Err(_) => LayerResult {
69 layer: "clamav",
70 verdict: LayerVerdict::Error,
71 detail: Some("ClamAV scan timed out".to_string()),
72 },
73 }
74 }
75
76 /// Streaming entry. Pumps `reader` into ClamAV INSTREAM frame-by-frame
77 /// rather than buffering the whole object first. Frame shape matches the
78 /// buffered path: 4-byte BE length + bytes, terminated with `0u32`.
79 pub async fn scan_with_clamav_stream<R>(socket_path: &str, reader: R) -> LayerResult
80 where
81 R: tokio::io::AsyncRead + Unpin,
82 {
83 let timeout = Duration::from_secs(constants::SCAN_CLAMAV_TIMEOUT_SECS);
84
85 match tokio::time::timeout(timeout, scan_instream_streaming(socket_path, reader)).await {
86 Ok(Ok(result)) => result,
87 Ok(Err(e)) => LayerResult {
88 layer: "clamav",
89 verdict: LayerVerdict::Error,
90 detail: Some(format!("ClamAV error: {e}")),
91 },
92 Err(_) => LayerResult {
93 layer: "clamav",
94 verdict: LayerVerdict::Error,
95 detail: Some("ClamAV scan timed out".to_string()),
96 },
97 }
98 }
99
100 async fn scan_instream(socket_path: &str, data: &[u8]) -> Result<LayerResult, String> {
101 let mut stream = UnixStream::connect(socket_path)
102 .await
103 .map_err(|e| format!("Failed to connect to ClamAV socket: {e}"))?;
104
105 // Send INSTREAM command
106 stream
107 .write_all(b"zINSTREAM\0")
108 .await
109 .map_err(|e| format!("Failed to send INSTREAM command: {e}"))?;
110
111 // Send data in chunks: each chunk prefixed with 4-byte big-endian length.
112 // If clamd drops the connection mid-stream (e.g. it hit StreamMaxLength), a
113 // write fails with EPIPE. Route that to the same fail-closed incomplete-scan
114 // hold the streaming path uses (`incomplete_after_clamd_dropped`) instead of
115 // returning an Err, an Err here was mapped to a clamav "degraded" error and
116 // then fail-OPEN, the opposite posture from the streaming twin (Run 15).
117 for chunk in data.chunks(CHUNK_SIZE) {
118 let len = (chunk.len() as u32).to_be_bytes();
119 if stream.write_all(&len).await.is_err() || stream.write_all(chunk).await.is_err() {
120 return Ok(incomplete_after_clamd_dropped(&mut stream).await);
121 }
122 }
123
124 // Send zero-length chunk to signal end of data
125 if stream.write_all(&0u32.to_be_bytes()).await.is_err() {
126 return Ok(incomplete_after_clamd_dropped(&mut stream).await);
127 }
128
129 // Read response (capped at 16 KB, verdicts are typically under 100 bytes)
130 let mut response = Vec::with_capacity(256);
131 stream
132 .take(16_384)
133 .read_to_end(&mut response)
134 .await
135 .map_err(|e| format!("Failed to read ClamAV response: {e}"))?;
136
137 // A 16 KB response means we hit the take() cap, clamd's verdict was
138 // longer than expected (typical verdicts are <100 bytes) and we don't
139 // know how much was truncated. Treat as suspicious / fail-closed rather
140 // than letting `parse_clamav_response` see a half-token and return Error
141 // (which under FailOpen would silently Pass). EICAR's "FOUND" suffix
142 // might be just beyond the cap; assume the worst.
143 if response.len() >= 16_384 {
144 return Ok(LayerResult {
145 layer: "clamav",
146 verdict: LayerVerdict::Fail,
147 detail: Some(
148 "ClamAV response truncated at 16 KB cap, refusing to interpret".to_string(),
149 ),
150 });
151 }
152
153 let response_str = String::from_utf8_lossy(&response);
154 let response_str = response_str.trim_end_matches('\0').trim();
155
156 Ok(clamav_layer_result(response_str))
157 }
158
159 async fn scan_instream_streaming<R>(socket_path: &str, mut reader: R) -> Result<LayerResult, String>
160 where
161 R: tokio::io::AsyncRead + Unpin,
162 {
163 let mut stream = UnixStream::connect(socket_path)
164 .await
165 .map_err(|e| format!("Failed to connect to ClamAV socket: {e}"))?;
166
167 stream
168 .write_all(b"zINSTREAM\0")
169 .await
170 .map_err(|e| format!("Failed to send INSTREAM command: {e}"))?;
171
172 // A write failure *after* the connection is established and the INSTREAM
173 // command was accepted means clamd dropped us mid-stream. The common cause
174 // is StreamMaxLength enforcement: clamd writes a size-limit ERROR and closes
175 // the socket, so our next write gets EPIPE. That is a reachable-but-
176 // incomplete scan (a coverage gap that must fail CLOSED), NOT the fail-open
177 // "daemon unreachable" bucket, so on any such write error we read clamd's
178 // pending verdict and classify it (a size-limit reply parses to
179 // NotFullyScanned → clamav_incomplete → FailClosed) rather than propagating
180 // a transport Err. This keeps the streaming path's size-limit handling
181 // identical to the buffered path's (CHRONIC S1 intent).
182 let mut buf = vec![0u8; CHUNK_SIZE];
183 loop {
184 let n = reader
185 .read(&mut buf)
186 .await
187 .map_err(|e| format!("Failed to read source: {e}"))?;
188 if n == 0 {
189 break;
190 }
191 let len = (n as u32).to_be_bytes();
192 if stream.write_all(&len).await.is_err() || stream.write_all(&buf[..n]).await.is_err() {
193 return Ok(incomplete_after_clamd_dropped(&mut stream).await);
194 }
195 }
196
197 if stream.write_all(&0u32.to_be_bytes()).await.is_err() {
198 return Ok(incomplete_after_clamd_dropped(&mut stream).await);
199 }
200
201 let mut response = Vec::with_capacity(256);
202 stream
203 .take(16_384)
204 .read_to_end(&mut response)
205 .await
206 .map_err(|e| format!("Failed to read ClamAV response: {e}"))?;
207
208 // A 16 KB response means we hit the take() cap, clamd's verdict was
209 // longer than expected (typical verdicts are <100 bytes) and we don't
210 // know how much was truncated. Treat as suspicious / fail-closed rather
211 // than letting `parse_clamav_response` see a half-token and return Error
212 // (which under FailOpen would silently Pass). EICAR's "FOUND" suffix
213 // might be just beyond the cap; assume the worst.
214 if response.len() >= 16_384 {
215 return Ok(LayerResult {
216 layer: "clamav",
217 verdict: LayerVerdict::Fail,
218 detail: Some(
219 "ClamAV response truncated at 16 KB cap, refusing to interpret".to_string(),
220 ),
221 });
222 }
223
224 let response_str = String::from_utf8_lossy(&response);
225 let response_str = response_str.trim_end_matches('\0').trim();
226
227 Ok(clamav_layer_result(response_str))
228 }
229
230 /// Classified outcome of a clamd INSTREAM reply we *successfully received*.
231 ///
232 /// The distinction this enum forces is the whole point of CHRONIC S1's fix:
233 /// transport failures (unreachable daemon / timeout) are handled in the scan
234 /// entry points and are legitimately FailOpen, clamd being down must not block
235 /// every upload. But by the time we are parsing a reply, **clamd is reachable**.
236 /// An error reply at that point (size/scan-size/recursion limit, or anything we
237 /// can't parse) means the file was *not cleanly scanned*, a coverage gap that
238 /// must fail CLOSED, never be conflated into the FailOpen unreachable bucket.
239 /// Because this is an exhaustive enum mapped at one place
240 /// ([`clamav_layer_result`]), a future reply kind can't silently fall into the
241 /// FailOpen path the way the old single `LayerVerdict::Error` return did.
242 #[derive(Debug, PartialEq, Eq)]
243 enum ClamavResponse {
244 Clean,
245 Infected(String),
246 /// clamd responded but produced no clean verdict, a limit was hit or the
247 /// reply was unparseable. The file was not fully scanned ⇒ fail closed.
248 NotFullyScanned(String),
249 }
250
251 /// Parse a clamd INSTREAM response string into a [`ClamavResponse`].
252 /// Extracted for testability, the socket/IO layer just feeds the string in.
253 ///
254 /// ClamAV response format:
255 /// - `"stream: OK"`, clean
256 /// - `"stream: <virus_name> FOUND"`, infected
257 /// - `"stream: <error> ERROR"` (e.g. `"INSTREAM size limit exceeded"`),
258 /// empty, or anything else, clamd is reachable but did not fully scan.
259 fn parse_clamav_response(response: &str) -> ClamavResponse {
260 if response == "stream: OK" {
261 ClamavResponse::Clean
262 } else if response.ends_with("FOUND") {
263 // Extract virus name: "stream: Eicar-Signature FOUND" → "Eicar-Signature"
264 let virus_name = response
265 .strip_prefix("stream: ")
266 .unwrap_or(response)
267 .strip_suffix(" FOUND")
268 .unwrap_or(response);
269 ClamavResponse::Infected(virus_name.to_string())
270 } else {
271 ClamavResponse::NotFullyScanned(response.to_string())
272 }
273 }
274
275 /// Map a *received* clamd reply to a [`LayerResult`]. A `NotFullyScanned`
276 /// outcome is emitted under the dedicated `clamav_incomplete` layer, whose
277 /// `error_policy_for` entry is **FailClosed**, so an un-scanned file is held for
278 /// review. A genuinely unreachable daemon / timeout is reported separately in
279 /// the scan entry points under the `clamav` layer (FailOpen). Routing the two
280 /// error kinds to different layer identities is the same mechanism the
281 /// `scan_panic` / `scan_size_limit` pseudo-layers already use to pick a policy.
282 fn clamav_layer_result(response: &str) -> LayerResult {
283 match parse_clamav_response(response) {
284 ClamavResponse::Clean => LayerResult {
285 layer: "clamav",
286 verdict: LayerVerdict::Pass,
287 detail: None,
288 },
289 ClamavResponse::Infected(name) => LayerResult {
290 layer: "clamav",
291 verdict: LayerVerdict::Fail,
292 detail: Some(format!("ClamAV detection: {name}")),
293 },
294 ClamavResponse::NotFullyScanned(resp) => LayerResult {
295 layer: "clamav_incomplete",
296 verdict: LayerVerdict::Error,
297 detail: Some(format!(
298 "ClamAV did not fully scan (held for review): {resp}"
299 )),
300 },
301 }
302 }
303
304 /// clamd dropped the connection mid-INSTREAM (commonly StreamMaxLength). Read
305 /// whatever verdict it managed to send before closing and classify it; a
306 /// size-limit reply parses to `NotFullyScanned` → fail-closed
307 /// `clamav_incomplete`. When nothing legible was sent, synthesize the same
308 /// fail-closed outcome. This never returns the fail-open `clamav` transport
309 /// bucket: clamd was reachable, the scan just didn't complete.
310 async fn incomplete_after_clamd_dropped(stream: &mut UnixStream) -> LayerResult {
311 let mut response = Vec::with_capacity(256);
312 let _ = (&mut *stream).take(16_384).read_to_end(&mut response).await;
313 let response_str = String::from_utf8_lossy(&response);
314 let response_str = response_str.trim_end_matches('\0').trim();
315 if response_str.is_empty() {
316 LayerResult {
317 layer: "clamav_incomplete",
318 verdict: LayerVerdict::Error,
319 detail: Some(
320 "ClamAV closed the connection mid-stream without a verdict \
321 (likely stream size limit); held for review"
322 .to_string(),
323 ),
324 }
325 } else {
326 clamav_layer_result(response_str)
327 }
328 }
329
330 #[cfg(test)]
331 mod tests {
332 use super::*;
333
334 // Clean responses
335
336 #[test]
337 fn clean_response_passes() {
338 assert_eq!(parse_clamav_response("stream: OK"), ClamavResponse::Clean);
339 let result = clamav_layer_result("stream: OK");
340 assert_eq!(result.verdict, LayerVerdict::Pass);
341 assert_eq!(result.layer, "clamav");
342 assert!(result.detail.is_none());
343 }
344
345 // Malware detections
346
347 #[test]
348 fn eicar_detection_fails() {
349 assert_eq!(
350 parse_clamav_response("stream: Eicar-Signature FOUND"),
351 ClamavResponse::Infected("Eicar-Signature".to_string())
352 );
353 let result = clamav_layer_result("stream: Eicar-Signature FOUND");
354 assert_eq!(result.verdict, LayerVerdict::Fail);
355 assert_eq!(result.layer, "clamav");
356 assert!(result.detail.unwrap().contains("Eicar-Signature"));
357 }
358
359 #[test]
360 fn complex_virus_name_extracted() {
361 let result = clamav_layer_result("stream: Win.Test.EICAR_HDB-1 FOUND");
362 assert_eq!(result.verdict, LayerVerdict::Fail);
363 assert!(result.detail.unwrap().contains("Win.Test.EICAR_HDB-1"));
364 }
365
366 #[test]
367 fn trojan_detection_fails() {
368 let result = clamav_layer_result("stream: Win.Trojan.Agent-123456 FOUND");
369 assert_eq!(result.verdict, LayerVerdict::Fail);
370 assert!(result.detail.unwrap().contains("Win.Trojan.Agent-123456"));
371 }
372
373 // Not-fully-scanned responses (clamd reachable, no clean verdict)
374 //
375 // CHRONIC S1: every one of these must FAIL CLOSED, not FailOpen. They are
376 // emitted under the `clamav_incomplete` layer (FailClosed in
377 // `error_policy_for`), distinct from the FailOpen `clamav` layer used for an
378 // unreachable daemon.
379
380 #[test]
381 fn empty_response_is_not_fully_scanned() {
382 assert_eq!(
383 parse_clamav_response(""),
384 ClamavResponse::NotFullyScanned(String::new())
385 );
386 assert_eq!(clamav_layer_result("").layer, "clamav_incomplete");
387 assert_eq!(clamav_layer_result("").verdict, LayerVerdict::Error);
388 }
389
390 #[test]
391 fn garbage_response_is_not_fully_scanned() {
392 let result = clamav_layer_result("this is not a valid response");
393 assert_eq!(result.layer, "clamav_incomplete");
394 assert_eq!(result.verdict, LayerVerdict::Error);
395 }
396
397 #[test]
398 fn size_limit_error_fails_closed_not_open() {
399 // The regression that defined CHRONIC S1: a payload sized past clamd's
400 // StreamMaxLength yields this reply on a HEALTHY clamd. It must route to
401 // the FailClosed `clamav_incomplete` layer, never the FailOpen `clamav`
402 // layer (which would skip → Clean).
403 let response = "stream: INSTREAM size limit exceeded ERROR";
404 assert!(matches!(
405 parse_clamav_response(response),
406 ClamavResponse::NotFullyScanned(_)
407 ));
408 let result = clamav_layer_result(response);
409 assert_eq!(result.layer, "clamav_incomplete");
410 assert_eq!(result.verdict, LayerVerdict::Error);
411 }
412 }
413