rama_http/layer/traffic_writer/
file.rs

1use super::{
2    RequestWriter, ResponseWriter, TrafficWriterId, WriterMode, ensure_traffic_writer_id,
3    write_headers_body_flags,
4};
5use crate::{Request, Response, io::write_http_request_streaming};
6use rama_core::error::BoxError;
7use rama_core::extensions::ExtensionsRef as _;
8use rama_core::telemetry::tracing;
9use rama_utils::fs::{CreatedFilePermissions, OpenOptions, is_reserved_device_name};
10use rama_utils::time::unix_timestamp_millis;
11use std::{
12    io,
13    path::{Path, PathBuf},
14    sync::Arc,
15};
16use tokio::io::{AsyncWriteExt as _, BufWriter};
17
18const MAX_PREFIX_LEN: usize = 32;
19const MAX_METHOD_LEN: usize = 16;
20const MAX_FILE_NAME_LEN: usize = 160;
21const CREATE_ATTEMPTS: usize = 8;
22
23/// Writes every captured request and response to an independent file.
24///
25/// Independent files allow bodies to stream concurrently without frame
26/// interleaving or cross-message backpressure. Generated filenames use only
27/// portable ASCII characters and are bounded to 160 bytes, comfortably below
28/// the commonly supported 255-byte component limit. The caller-provided
29/// destination can still be rejected by the operating system when the complete
30/// path exceeds a platform-specific path limit.
31/// Raw URIs are never included because they can contain secrets, path
32/// separators, platform-specific characters, or unbounded input.
33/// Request and response files captured for the same HTTP exchange contain the
34/// same UUID, regardless of request/response writer layer order.
35///
36/// Files are created without overwriting existing paths. On Unix, newly
37/// created files receive owner-only read/write permissions before the process
38/// umask is applied. The destination directory is canonicalized and every file
39/// open is confined to it.
40#[derive(Clone, Debug)]
41pub struct PerMessageFileWriter {
42    directory: Arc<PathBuf>,
43    prefix: Arc<str>,
44    request_mode: Option<WriterMode>,
45    response_mode: Option<WriterMode>,
46}
47
48impl PerMessageFileWriter {
49    /// Create a per-message file writer, creating `directory` when necessary.
50    ///
51    /// `prefix` must be 1 to 32 ASCII letters, digits, `-`, or `_`. This
52    /// deliberately conservative alphabet is valid in filenames on Unix,
53    /// Windows, and common removable filesystems.
54    pub async fn try_new(
55        directory: impl Into<PathBuf>,
56        prefix: impl AsRef<str>,
57    ) -> io::Result<Self> {
58        let prefix = validate_prefix(prefix.as_ref())?;
59        let directory = directory.into();
60        tokio::fs::create_dir_all(&directory).await?;
61        let directory = tokio::fs::canonicalize(directory).await?;
62        Ok(Self {
63            directory: Arc::new(directory),
64            prefix: Arc::from(prefix),
65            request_mode: Some(WriterMode::All),
66            response_mode: Some(WriterMode::All),
67        })
68    }
69
70    /// Set how request files are written.
71    #[must_use]
72    pub fn with_request_mode(mut self, mode: Option<WriterMode>) -> Self {
73        self.request_mode = mode;
74        self
75    }
76
77    /// Set how response files are written.
78    #[must_use]
79    pub fn with_response_mode(mut self, mode: Option<WriterMode>) -> Self {
80        self.response_mode = mode;
81        self
82    }
83
84    /// Return the canonical destination directory.
85    #[must_use]
86    pub fn directory(&self) -> &Path {
87        self.directory.as_ref()
88    }
89
90    /// Return the portable filename prefix.
91    #[must_use]
92    pub fn prefix(&self) -> &str {
93        &self.prefix
94    }
95
96    /// Return the configured request writer mode.
97    #[must_use]
98    pub const fn request_mode(&self) -> Option<WriterMode> {
99        self.request_mode
100    }
101
102    /// Return the configured response writer mode.
103    #[must_use]
104    pub const fn response_mode(&self) -> Option<WriterMode> {
105        self.response_mode
106    }
107
108    async fn open_unique(
109        &self,
110        name: impl Fn(usize) -> String,
111    ) -> io::Result<(BufWriter<tokio::fs::File>, PathBuf)> {
112        for attempt in 0..CREATE_ATTEMPTS {
113            let filename = name(attempt);
114            if !is_portable_filename(&filename) {
115                return Err(io::Error::new(
116                    io::ErrorKind::InvalidInput,
117                    "generated traffic capture filename is not portable",
118                ));
119            }
120
121            let result = OpenOptions::new()
122                .write(true)
123                .create_new(true)
124                .created_file_permissions(CreatedFilePermissions::OwnerReadWrite)
125                .jail(self.directory.as_ref())
126                .open(&filename)
127                .await;
128            match result {
129                Ok(file) => {
130                    let path = self.directory.join(filename);
131                    return Ok((BufWriter::new(file), path));
132                }
133                Err(error) if error.kind() == io::ErrorKind::AlreadyExists => {}
134                Err(error) => return Err(error),
135            }
136        }
137
138        Err(io::Error::new(
139            io::ErrorKind::AlreadyExists,
140            "failed to allocate a unique traffic capture filename",
141        ))
142    }
143}
144
145impl RequestWriter for PerMessageFileWriter {
146    async fn write_request(&self, request: Request) {
147        let (write_headers, write_body) = write_headers_body_flags(self.request_mode);
148        if !write_headers && !write_body {
149            return;
150        }
151
152        let method = portable_method(request.method().as_str());
153        let traffic_writer_id = ensure_traffic_writer_id(request.extensions());
154        let timestamp = unix_timestamp_millis();
155        let opened = self
156            .open_unique(|attempt| {
157                request_filename(&self.prefix, timestamp, &method, traffic_writer_id, attempt)
158            })
159            .await;
160        let (mut file, path) = match opened {
161            Ok(value) => value,
162            Err(error) => {
163                tracing::error!(%error, "failed to create request capture file");
164                return;
165            }
166        };
167
168        let result =
169            write_http_request_streaming(&mut file, request, write_headers, write_body).await;
170        finish_file_write(&mut file, &path, "request", result).await;
171    }
172}
173
174impl ResponseWriter for PerMessageFileWriter {
175    async fn write_response(&self, response: Response) {
176        let (write_headers, write_body) = write_headers_body_flags(self.response_mode);
177        if !write_headers && !write_body {
178            return;
179        }
180
181        let status = response.status().as_u16();
182        let traffic_writer_id = ensure_traffic_writer_id(response.extensions());
183        let timestamp = unix_timestamp_millis();
184        let opened = self
185            .open_unique(|attempt| {
186                response_filename(&self.prefix, timestamp, status, traffic_writer_id, attempt)
187            })
188            .await;
189        let (mut file, path) = match opened {
190            Ok(value) => value,
191            Err(error) => {
192                tracing::error!(%error, "failed to create response capture file");
193                return;
194            }
195        };
196
197        let result = crate::io::write_http_response_streaming(
198            &mut file,
199            response,
200            write_headers,
201            write_body,
202        )
203        .await;
204        finish_file_write(&mut file, &path, "response", result).await;
205    }
206}
207
208async fn finish_file_write(
209    file: &mut BufWriter<tokio::fs::File>,
210    path: &Path,
211    message_kind: &'static str,
212    write_result: Result<(), BoxError>,
213) {
214    if let Err(error) = write_result {
215        tracing::error!(
216            %error,
217            path = %path.display(),
218            message_kind,
219            "failed to write HTTP capture file"
220        );
221    }
222    if let Err(error) = file.flush().await {
223        tracing::error!(
224            %error,
225            path = %path.display(),
226            message_kind,
227            "failed to flush HTTP capture file"
228        );
229    }
230}
231
232fn validate_prefix(prefix: &str) -> io::Result<String> {
233    if prefix.is_empty()
234        || prefix.len() > MAX_PREFIX_LEN
235        || !prefix
236            .bytes()
237            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_'))
238        || is_reserved_device_name(prefix)
239    {
240        return Err(io::Error::new(
241            io::ErrorKind::InvalidInput,
242            "traffic capture filename prefix must be 1 to 32 portable ASCII letters, digits, '-' or '_' and not a reserved device name",
243        ));
244    }
245    Ok(prefix.to_owned())
246}
247
248fn portable_method(method: &str) -> String {
249    method
250        .bytes()
251        .take(MAX_METHOD_LEN)
252        .map(|byte| {
253            if byte.is_ascii_alphanumeric() {
254                char::from(byte.to_ascii_uppercase())
255            } else {
256                '_'
257            }
258        })
259        .collect()
260}
261
262fn request_filename(
263    prefix: &str,
264    timestamp: i64,
265    method: &str,
266    traffic_writer_id: TrafficWriterId,
267    attempt: usize,
268) -> String {
269    let id = traffic_writer_id.0.simple();
270    if attempt == 0 {
271        format!("{prefix}-{timestamp}-request-{method}-{id}.http")
272    } else {
273        format!("{prefix}-{timestamp}-request-{method}-{id}-{attempt}.http")
274    }
275}
276
277fn response_filename(
278    prefix: &str,
279    timestamp: i64,
280    status: u16,
281    traffic_writer_id: TrafficWriterId,
282    attempt: usize,
283) -> String {
284    let id = traffic_writer_id.0.simple();
285    if attempt == 0 {
286        format!("{prefix}-{timestamp}-response-{status}-{id}.http")
287    } else {
288        format!("{prefix}-{timestamp}-response-{status}-{id}-{attempt}.http")
289    }
290}
291
292fn is_portable_filename(filename: &str) -> bool {
293    !filename.is_empty()
294        && filename.len() <= MAX_FILE_NAME_LEN
295        && filename
296            .bytes()
297            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.'))
298        && !filename.ends_with(['.', ' '])
299        && !is_reserved_device_name(filename)
300}
301
302#[cfg(test)]
303mod tests {
304    use super::*;
305    use crate::{
306        Body, Method, StatusCode,
307        body::util::BodyExt as _,
308        layer::traffic_writer::{RequestWriterLayer, ResponseWriterLayer},
309    };
310    use ahash::{HashSet, HashSetExt as _};
311    use rama_core::{Layer as _, Service as _, rt::Executor, service::service_fn};
312    use std::{convert::Infallible, time::Duration};
313    use uuid::Uuid;
314
315    async fn wait_for_exchange_files(directory: &Path) -> Vec<(String, Vec<u8>)> {
316        tokio::time::timeout(Duration::from_secs(1), async {
317            loop {
318                let mut entries = tokio::fs::read_dir(directory).await.unwrap();
319                let mut files = Vec::new();
320                while let Some(entry) = entries.next_entry().await.unwrap() {
321                    files.push((
322                        entry.file_name().into_string().unwrap(),
323                        tokio::fs::read(entry.path()).await.unwrap(),
324                    ));
325                }
326                if files.len() == 2
327                    && files
328                        .iter()
329                        .any(|(_, bytes)| bytes.ends_with(b"\r\nrequest-body"))
330                    && files
331                        .iter()
332                        .any(|(_, bytes)| bytes.ends_with(b"\r\nresponse-body"))
333                {
334                    return files;
335                }
336                tokio::time::sleep(Duration::from_millis(1)).await;
337            }
338        })
339        .await
340        .expect("request and response capture files should finish")
341    }
342
343    fn assert_exchange_files_share_id(files: &[(String, Vec<u8>)]) {
344        fn id(name: &str) -> &str {
345            name.strip_suffix(".http")
346                .unwrap()
347                .rsplit('-')
348                .next()
349                .unwrap()
350        }
351
352        let request = files
353            .iter()
354            .find(|(name, _)| name.contains("-request-POST-"))
355            .unwrap();
356        let response = files
357            .iter()
358            .find(|(name, _)| name.contains("-response-200-"))
359            .unwrap();
360        let request_id = id(&request.0);
361        let response_id = id(&response.0);
362        assert_eq!(request_id.len(), 32);
363        assert!(request_id.bytes().all(|byte| byte.is_ascii_hexdigit()));
364        assert_eq!(request_id, response_id);
365    }
366
367    async fn consume_request(request: Request) -> Result<Response, Infallible> {
368        assert_eq!(
369            request.into_body().collect().await.unwrap().to_bytes(),
370            "request-body"
371        );
372        Ok(Response::new(Body::from("response-body")))
373    }
374
375    #[test]
376    fn generated_filenames_are_unique_portable_and_bounded() {
377        let method = portable_method("CUSTOM/METHOD-THAT-IS-TOO-LONG");
378        assert_eq!(method, "CUSTOM_METHOD_TH");
379        let mut names = HashSet::new();
380
381        for _ in 0..128 {
382            let traffic_writer_id = TrafficWriterId(Uuid::new_v4());
383            let request = request_filename(
384                "portable_prefix-123456789012345",
385                1_700_000_000_000,
386                &method,
387                traffic_writer_id,
388                0,
389            );
390            let response = response_filename(
391                "portable_prefix-123456789012345",
392                1_700_000_000_001,
393                599,
394                traffic_writer_id,
395                0,
396            );
397            let id = traffic_writer_id.0.simple().to_string();
398            assert!(request.contains(&id));
399            assert!(response.contains(&id));
400            for filename in [request, response] {
401                assert!(is_portable_filename(&filename));
402                assert!(filename.len() <= MAX_FILE_NAME_LEN);
403                assert_eq!(Path::new(&filename).components().count(), 1);
404                assert!(!filename.contains("access_token"));
405                assert!(names.insert(filename));
406            }
407        }
408
409        let traffic_writer_id = TrafficWriterId(Uuid::new_v4());
410        let base = request_filename("capture", 1, "GET", traffic_writer_id, 0);
411        let retry = request_filename("capture", 1, "GET", traffic_writer_id, 1);
412        assert_ne!(base, retry);
413        assert!(retry.ends_with(&format!("-{}-1.http", traffic_writer_id.0.simple())));
414        assert!(is_portable_filename(&retry));
415
416        let base = response_filename("capture", 1, 200, traffic_writer_id, 0);
417        let retry = response_filename("capture", 1, 200, traffic_writer_id, 1);
418        assert_ne!(base, retry);
419        assert!(retry.ends_with(&format!("-{}-1.http", traffic_writer_id.0.simple())));
420        assert!(is_portable_filename(&retry));
421
422        let longest = request_filename(
423            "12345678901234567890123456789012",
424            i64::MIN,
425            &method,
426            traffic_writer_id,
427            CREATE_ATTEMPTS - 1,
428        );
429        assert!(longest.len() <= MAX_FILE_NAME_LEN);
430        assert!(is_portable_filename(&longest));
431    }
432
433    #[tokio::test]
434    async fn file_layers_share_exchange_id_with_request_layer_outermost() {
435        let temp = tempfile::tempdir().unwrap();
436        let executor = Executor::new();
437        let request = RequestWriterLayer::file_per_request(
438            &executor,
439            temp.path(),
440            "traffic",
441            Some(WriterMode::All),
442        )
443        .await
444        .unwrap();
445        let response = ResponseWriterLayer::file_per_response(
446            &executor,
447            temp.path(),
448            "traffic",
449            Some(WriterMode::All),
450        )
451        .await
452        .unwrap();
453        let service = request.into_layer(response.into_layer(service_fn(consume_request)));
454
455        let response = service
456            .serve(
457                Request::builder()
458                    .method(Method::POST)
459                    .uri("/capture")
460                    .body(Body::from("request-body"))
461                    .unwrap(),
462            )
463            .await
464            .unwrap();
465        assert_eq!(
466            response.into_body().collect().await.unwrap().to_bytes(),
467            "response-body"
468        );
469        drop(service);
470
471        let files = wait_for_exchange_files(temp.path()).await;
472        assert_exchange_files_share_id(&files);
473    }
474
475    #[tokio::test]
476    async fn file_layers_share_exchange_id_with_response_layer_outermost() {
477        let temp = tempfile::tempdir().unwrap();
478        let executor = Executor::new();
479        let request = RequestWriterLayer::file_per_request(
480            &executor,
481            temp.path(),
482            "traffic",
483            Some(WriterMode::All),
484        )
485        .await
486        .unwrap();
487        let response = ResponseWriterLayer::file_per_response(
488            &executor,
489            temp.path(),
490            "traffic",
491            Some(WriterMode::All),
492        )
493        .await
494        .unwrap();
495        let service = response.into_layer(request.into_layer(service_fn(consume_request)));
496
497        let response = service
498            .serve(
499                Request::builder()
500                    .method(Method::POST)
501                    .uri("/capture")
502                    .body(Body::from("request-body"))
503                    .unwrap(),
504            )
505            .await
506            .unwrap();
507        assert_eq!(
508            response.into_body().collect().await.unwrap().to_bytes(),
509            "response-body"
510        );
511        drop(service);
512
513        let files = wait_for_exchange_files(temp.path()).await;
514        assert_exchange_files_share_id(&files);
515    }
516
517    #[tokio::test]
518    async fn prefix_validation_is_platform_conservative() {
519        let temp = tempfile::tempdir().unwrap();
520        for prefix in [
521            "",
522            "contains space",
523            "contains.dot",
524            "../escape",
525            r"back\slash",
526            "unicode-λ",
527            "CON",
528            "nul",
529            "123456789012345678901234567890123",
530        ] {
531            let error = PerMessageFileWriter::try_new(temp.path(), prefix)
532                .await
533                .unwrap_err();
534            assert_eq!(error.kind(), io::ErrorKind::InvalidInput, "{prefix:?}");
535        }
536
537        let writer = PerMessageFileWriter::try_new(temp.path(), "Abc_123-xyz-12345678901234567890")
538            .await
539            .unwrap();
540        assert_eq!(writer.prefix().len(), MAX_PREFIX_LEN);
541        assert_eq!(writer.directory(), temp.path().canonicalize().unwrap());
542    }
543
544    #[tokio::test]
545    async fn unique_creation_never_overwrites_an_existing_file() {
546        let temp = tempfile::tempdir().unwrap();
547        let writer = PerMessageFileWriter::try_new(temp.path(), "capture")
548            .await
549            .unwrap();
550        let colliding = "capture-1-request-GET-00000000000000000000000000000000.http";
551        let unique = "capture-2-request-GET-00000000000000000000000000000000.http";
552        assert!(is_portable_filename(colliding));
553        assert!(is_portable_filename(unique));
554        let path = temp.path().join(colliding);
555        tokio::fs::write(&path, b"existing").await.unwrap();
556
557        let (_, created_path) = writer
558            .open_unique(|attempt| {
559                if attempt == 0 {
560                    colliding.to_owned()
561                } else {
562                    unique.to_owned()
563                }
564            })
565            .await
566            .unwrap();
567        assert_eq!(created_path, writer.directory().join(unique));
568        assert_eq!(tokio::fs::read(path).await.unwrap(), b"existing");
569    }
570
571    #[tokio::test]
572    async fn unique_creation_reports_exhausted_collisions() {
573        let temp = tempfile::tempdir().unwrap();
574        let writer = PerMessageFileWriter::try_new(temp.path(), "capture")
575            .await
576            .unwrap();
577        let filename = "capture-1-request-GET-00000000000000000000000000000000.http";
578        tokio::fs::write(temp.path().join(filename), b"existing")
579            .await
580            .unwrap();
581
582        let attempts = std::sync::atomic::AtomicUsize::new(0);
583        let error = writer
584            .open_unique(|_| {
585                attempts.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
586                filename.to_owned()
587            })
588            .await
589            .unwrap_err();
590        assert_eq!(error.kind(), io::ErrorKind::AlreadyExists);
591        assert_eq!(
592            attempts.load(std::sync::atomic::Ordering::Relaxed),
593            CREATE_ATTEMPTS
594        );
595    }
596
597    #[tokio::test]
598    async fn unique_creation_propagates_non_collision_errors() {
599        let temp = tempfile::tempdir().unwrap();
600        let directory = temp.path().join("removed");
601        let writer = PerMessageFileWriter::try_new(&directory, "capture")
602            .await
603            .unwrap();
604        tokio::fs::remove_dir(directory).await.unwrap();
605
606        let error = writer
607            .open_unique(|_| "capture-unique.http".to_owned())
608            .await
609            .unwrap_err();
610        assert_eq!(error.kind(), io::ErrorKind::NotFound);
611    }
612
613    #[tokio::test]
614    async fn unique_creation_rejects_non_portable_generated_names() {
615        let temp = tempfile::tempdir().unwrap();
616        let writer = PerMessageFileWriter::try_new(temp.path(), "capture")
617            .await
618            .unwrap();
619
620        let overly_long = "x".repeat(MAX_FILE_NAME_LEN + 1);
621        for filename in ["../escape.http", "trailing.", overly_long.as_str()] {
622            let error = writer
623                .open_unique(|_| filename.to_owned())
624                .await
625                .unwrap_err();
626            assert_eq!(error.kind(), io::ErrorKind::InvalidInput);
627        }
628        assert!(
629            tokio::fs::read_dir(temp.path())
630                .await
631                .unwrap()
632                .next_entry()
633                .await
634                .unwrap()
635                .is_none()
636        );
637    }
638
639    #[tokio::test]
640    async fn request_and_response_files_stream_independently() {
641        let temp = tempfile::tempdir().unwrap();
642        let writer = PerMessageFileWriter::try_new(temp.path(), "traffic")
643            .await
644            .unwrap();
645        let traffic_writer_id = TrafficWriterId(Uuid::new_v4());
646
647        let (first_tx, first_rx) = tokio::sync::mpsc::unbounded_channel();
648        let first_body = Body::from_stream(rama_core::futures::stream::unfold(
649            first_rx,
650            |mut receiver| async move { receiver.recv().await.map(|item| (item, receiver)) },
651        ));
652        let first_response = Response::builder()
653            .status(StatusCode::OK)
654            .body(first_body)
655            .unwrap();
656        first_response.extensions().insert(traffic_writer_id);
657        let first_writer = writer.clone();
658        let first = tokio::spawn(async move {
659            first_writer.write_response(first_response).await;
660        });
661        let second_request = Request::builder()
662            .method(Method::POST)
663            .uri("https://example.com/upload?token=secret")
664            .header("x-test", "yes")
665            .body(Body::from("second"))
666            .unwrap();
667        second_request.extensions().insert(traffic_writer_id);
668        let second = writer.write_request(second_request);
669        first_tx
670            .send(Ok::<_, std::convert::Infallible>(
671                rama_core::bytes::Bytes::from_static(b"first"),
672            ))
673            .unwrap();
674
675        tokio::time::timeout(std::time::Duration::from_secs(1), second)
676            .await
677            .expect("a separate request file must not wait for a live response");
678        drop(first_tx);
679        first.await.unwrap();
680
681        let mut entries = tokio::fs::read_dir(temp.path()).await.unwrap();
682        let mut files = Vec::new();
683        while let Some(entry) = entries.next_entry().await.unwrap() {
684            let name = entry.file_name().into_string().unwrap();
685            assert!(is_portable_filename(&name));
686            files.push((name, tokio::fs::read(entry.path()).await.unwrap()));
687        }
688        assert_eq!(files.len(), 2);
689        assert!(files.iter().any(|(name, bytes)| {
690            name.contains("-request-POST-")
691                && bytes.starts_with(b"POST /upload?token=secret HTTP/1.1\r\n")
692                && bytes.ends_with(b"\r\nsecond")
693        }));
694        assert!(files.iter().any(|(name, bytes)| {
695            name.contains("-response-200-")
696                && bytes.starts_with(b"HTTP/1.1 200 OK\r\n")
697                && bytes.ends_with(b"\r\nfirst")
698        }));
699        assert!(files.iter().all(|(name, _)| !name.contains("secret")));
700        let id = traffic_writer_id.0.simple().to_string();
701        assert!(files.iter().all(|(name, _)| name.contains(&id)));
702    }
703
704    #[tokio::test]
705    async fn body_error_flushes_partial_request_capture() {
706        let temp = tempfile::tempdir().unwrap();
707        let writer = PerMessageFileWriter::try_new(temp.path(), "traffic")
708            .await
709            .unwrap()
710            .with_response_mode(None);
711        let body = Body::from_stream(rama_core::futures::stream::iter([
712            Ok::<_, io::Error>(rama_core::bytes::Bytes::from_static(b"first")),
713            Err(io::Error::other("body failed")),
714        ]));
715
716        writer
717            .write_request(
718                Request::builder()
719                    .method(Method::POST)
720                    .uri("/broken")
721                    .body(body)
722                    .unwrap(),
723            )
724            .await;
725
726        let entry = tokio::fs::read_dir(temp.path())
727            .await
728            .unwrap()
729            .next_entry()
730            .await
731            .unwrap()
732            .unwrap();
733        assert_eq!(
734            tokio::fs::read(entry.path()).await.unwrap(),
735            b"POST /broken HTTP/1.1\r\n\r\nfirst"
736        );
737    }
738
739    #[tokio::test]
740    async fn disabled_request_does_not_create_a_file() {
741        let temp = tempfile::tempdir().unwrap();
742        let writer = PerMessageFileWriter::try_new(temp.path(), "traffic")
743            .await
744            .unwrap()
745            .with_request_mode(None);
746        writer
747            .write_request(Request::new(Body::from("ignored")))
748            .await;
749        assert!(
750            tokio::fs::read_dir(temp.path())
751                .await
752                .unwrap()
753                .next_entry()
754                .await
755                .unwrap()
756                .is_none()
757        );
758    }
759
760    #[tokio::test]
761    async fn disabled_response_does_not_create_a_file() {
762        let temp = tempfile::tempdir().unwrap();
763        let writer = PerMessageFileWriter::try_new(temp.path(), "traffic")
764            .await
765            .unwrap()
766            .with_response_mode(None);
767        writer
768            .write_response(Response::new(Body::from("ignored")))
769            .await;
770        assert!(
771            tokio::fs::read_dir(temp.path())
772                .await
773                .unwrap()
774                .next_entry()
775                .await
776                .unwrap()
777                .is_none()
778        );
779    }
780
781    #[tokio::test]
782    async fn response_headers_mode_creates_a_file() {
783        let temp = tempfile::tempdir().unwrap();
784        let writer = PerMessageFileWriter::try_new(temp.path(), "traffic")
785            .await
786            .unwrap()
787            .with_request_mode(None)
788            .with_response_mode(Some(WriterMode::Headers));
789        writer
790            .write_response(
791                Response::builder()
792                    .status(StatusCode::ACCEPTED)
793                    .header("x-mode", "headers")
794                    .body(Body::from("ignored"))
795                    .unwrap(),
796            )
797            .await;
798
799        let entry = tokio::fs::read_dir(temp.path())
800            .await
801            .unwrap()
802            .next_entry()
803            .await
804            .unwrap()
805            .unwrap();
806        let bytes = tokio::fs::read(entry.path()).await.unwrap();
807        assert!(bytes.starts_with(b"HTTP/1.1 202 Accepted\r\n"));
808        assert!(bytes.ends_with(b"x-mode: headers\r\n"));
809        assert!(!bytes.windows(7).any(|window| window == b"ignored"));
810    }
811
812    #[tokio::test]
813    async fn request_and_response_modes_control_file_contents() {
814        let temp = tempfile::tempdir().unwrap();
815        let writer = PerMessageFileWriter::try_new(temp.path(), "modes")
816            .await
817            .unwrap()
818            .with_request_mode(Some(WriterMode::Headers))
819            .with_response_mode(Some(WriterMode::Body));
820
821        writer
822            .write_request(
823                Request::builder()
824                    .uri("/headers")
825                    .header("x-mode", "headers")
826                    .body(Body::from("not written"))
827                    .unwrap(),
828            )
829            .await;
830        writer
831            .write_response(
832                Response::builder()
833                    .status(StatusCode::CREATED)
834                    .header("x-mode", "not written")
835                    .body(Body::from("body only"))
836                    .unwrap(),
837            )
838            .await;
839
840        let mut entries = tokio::fs::read_dir(temp.path()).await.unwrap();
841        let mut files = Vec::new();
842        while let Some(entry) = entries.next_entry().await.unwrap() {
843            files.push((
844                entry.file_name().into_string().unwrap(),
845                tokio::fs::read(entry.path()).await.unwrap(),
846            ));
847        }
848        assert_eq!(files.len(), 2);
849        assert!(files.iter().any(|(name, bytes)| {
850            name.contains("-request-GET-")
851                && bytes.starts_with(b"GET /headers HTTP/1.1\r\n")
852                && bytes.ends_with(b"x-mode: headers\r\n")
853                && !bytes.windows(11).any(|window| window == b"not written")
854        }));
855        assert!(
856            files.iter().any(|(name, bytes)| {
857                name.contains("-response-201-") && bytes == b"\r\nbody only"
858            })
859        );
860    }
861
862    #[cfg(unix)]
863    #[tokio::test]
864    async fn capture_files_are_not_group_or_world_accessible() {
865        use std::os::unix::fs::PermissionsExt as _;
866
867        let temp = tempfile::tempdir().unwrap();
868        let writer = PerMessageFileWriter::try_new(temp.path(), "traffic")
869            .await
870            .unwrap();
871        writer
872            .write_response(Response::new(Body::from("private")))
873            .await;
874
875        let entry = tokio::fs::read_dir(temp.path())
876            .await
877            .unwrap()
878            .next_entry()
879            .await
880            .unwrap()
881            .unwrap();
882        let mode = entry.metadata().await.unwrap().permissions().mode();
883        assert_eq!(mode & 0o077, 0);
884    }
885}
rama_http/layer/traffic_writer/file.rs

rama_http/layer/traffic_writer/
file.rs
