tor_hsservice/
replay.rs

1//! Facility for detecting and preventing replays on introduction requests.
2//!
3//! If we were to permit the introduction point to replay the same request
4//! multiple times, it would cause the service to contact the rendezvous point
5//! again with the same rendezvous cookie as before, which could help with
6//! traffic analysis.
7//!
8//! (This could also be a DoS vector if the introduction point decided to
9//! overload the service.)
10//!
11//! Because we use the same introduction point keys across restarts, we need to
12//! make sure that our replay logs are already persistent.  We do this by using
13//! a file on disk.
14
15mod ipt;
16#[cfg(feature = "hs-pow-full")]
17mod pow;
18
19use crate::internal_prelude::*;
20
21/// A probabilistic data structure to record fingerprints of observed Introduce2
22/// messages.
23///
24/// We need to record these fingerprints to prevent replay attacks; see the
25/// module documentation for an explanation of why that would be bad.
26///
27/// A ReplayLog should correspond to a `KP_hss_ntor` key, and should have the
28/// same lifespan: dropping it sooner will enable replays, but dropping it later
29/// will waste disk and memory.
30///
31/// False positives are allowed, to conserve on space.
32pub(crate) struct ReplayLog<T> {
33    /// The inner probabilistic data structure.
34    seen: data::Filter,
35    /// Persistent state file etc., if we're persistent
36    ///
37    /// If is is `None`, this RelayLog is ephemeral.
38    file: Option<PersistFile>,
39    /// [`PhantomData`] so rustc doesn't complain about the unused type param.
40    ///
41    /// This type represents the type of data that we're storing, as well as the type of the
42    /// key/name for that data.
43    replay_log_type: PhantomData<T>,
44}
45
46/// A [`ReplayLog`] for [`Introduce2`](tor_cell::relaycell::msg::Introduce2) messages.
47pub(crate) type IptReplayLog = ReplayLog<ipt::IptReplayLogType>;
48
49/// A [`ReplayLog`] for Proof-of-Work [`Nonce`](tor_hscrypto::pow::v1::Nonce)s.
50#[cfg(feature = "hs-pow-full")]
51pub(crate) type PowNonceReplayLog = ReplayLog<pow::PowNonceReplayLogType>;
52
53/// The length of the [`ReplayLogType::MAGIC`] constant.
54///
55// TODO: If Rust's constant expressions supported generics we wouldn't need this at all.
56const MAGIC_LEN: usize = 32;
57
58/// The length of the message that we store on disk, in bytes.
59///
60/// If the message is longer than this, then we will need to hash or truncate it before storing it
61/// to disk.
62///
63// TODO: Once const generics are good, this should be a associated constant for ReplayLogType.
64pub(crate) const OUTPUT_LEN: usize = 16;
65
66/// A trait to represent a set of types that ReplayLog can be used with.
67pub(crate) trait ReplayLogType {
68    // TODO: It would be nice to encode the directory name as a associated constant here, rather
69    // than having the external code pass it in to us.
70
71    /// The name of this item, used for the log filename.
72    type Name;
73
74    /// The type of the messages that we are ensuring the uniqueness of.
75    type Message;
76
77    /// A magic string that we put at the start of each log file, to make sure that
78    /// we don't confuse this file format with others.
79    const MAGIC: &'static [u8; MAGIC_LEN];
80
81    /// Convert [`Self::Name`] to a [`String`]
82    fn format_filename(name: &Self::Name) -> String;
83
84    /// Convert [`Self::Message`] to bytes that will be stored in the log.
85    fn transform_message(message: &Self::Message) -> [u8; OUTPUT_LEN];
86
87    /// Parse a filename into [`Self::Name`].
88    fn parse_log_leafname(leaf: &OsStr) -> Result<Self::Name, Cow<'static, str>>;
89}
90
91/// Persistent state file, and associated data
92///
93/// Stored as `ReplayLog.file`.
94#[derive(Debug)]
95pub(crate) struct PersistFile {
96    /// A file logging fingerprints of the messages we have seen.
97    file: BufWriter<File>,
98    /// Whether we had a possible partial write
99    ///
100    /// See the comment inside [`ReplayLog::check_for_replay`].
101    /// `Ok` means all is well.
102    /// `Err` means we may have written partial data to the actual file,
103    /// and need to make sure we're back at a record boundary.
104    needs_resynch: Result<(), ()>,
105    /// Filesystem lock which must not be released until after we finish writing
106    ///
107    /// Must come last so that the drop order is correct
108    #[allow(dead_code)] // Held just so we unlock on drop
109    lock: Arc<LockFileGuard>,
110}
111
112/// Replay log files have a `.bin` suffix.
113///
114/// The name of the file is determined by [`ReplayLogType::format_filename`].
115const REPLAY_LOG_SUFFIX: &str = ".bin";
116
117impl<T: ReplayLogType> ReplayLog<T> {
118    /// Create a new ReplayLog not backed by any data storage.
119    #[allow(dead_code)] // TODO #1186 Remove once something uses ReplayLog.
120    pub(crate) fn new_ephemeral() -> Self {
121        Self {
122            seen: data::Filter::new(),
123            file: None,
124            replay_log_type: PhantomData,
125        }
126    }
127
128    /// Create a ReplayLog backed by the file at a given path.
129    ///
130    /// If the file already exists, load its contents and append any new
131    /// contents to it; otherwise, create the file.
132    ///
133    /// **`lock` must already have been locked** and this
134    /// *cannot be assured by the type system*.
135    ///
136    /// # Limitations
137    ///
138    /// It is the caller's responsibility to make sure that there are never two
139    /// `ReplayLogs` open at once for the same path, or for two paths that
140    /// resolve to the same file.
141    pub(crate) fn new_logged(
142        dir: &InstanceRawSubdir,
143        name: &T::Name,
144    ) -> Result<Self, CreateIptError> {
145        let leaf = T::format_filename(name);
146        let path = dir.as_path().join(leaf);
147        let lock_guard = dir.raw_lock_guard();
148
149        Self::new_logged_inner(&path, lock_guard).map_err(|error| CreateIptError::OpenReplayLog {
150            file: path,
151            error: error.into(),
152        })
153    }
154
155    /// Inner function for `new_logged`, with reified arguments and raw error type
156    fn new_logged_inner(path: impl AsRef<Path>, lock: Arc<LockFileGuard>) -> io::Result<Self> {
157        let mut file = {
158            let mut options = OpenOptions::new();
159            options.read(true).write(true).create(true);
160
161            #[cfg(target_family = "unix")]
162            {
163                use std::os::unix::fs::OpenOptionsExt as _;
164                options.mode(0o600);
165            }
166
167            options.open(path)?
168        };
169
170        // If the file is new, we need to write the magic string. Else we must
171        // read it.
172        let file_len = file.metadata()?.len();
173        if file_len == 0 {
174            file.write_all(T::MAGIC)?;
175        } else {
176            let mut m = [0_u8; MAGIC_LEN];
177            file.read_exact(&mut m)?;
178            if &m != T::MAGIC {
179                return Err(io::Error::new(
180                    io::ErrorKind::InvalidData,
181                    LogContentError::UnrecognizedFormat,
182                ));
183            }
184
185            Self::truncate_to_multiple(&mut file, file_len)?;
186        }
187
188        // Now read the rest of the file.
189        let mut seen = data::Filter::new();
190        let mut r = BufReader::new(file);
191        loop {
192            let mut msg = [0_u8; OUTPUT_LEN];
193            match r.read_exact(&mut msg) {
194                Ok(()) => {
195                    let _ = seen.test_and_add(&msg); // ignore error.
196                }
197                Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => break,
198                Err(e) => return Err(e),
199            }
200        }
201        let mut file = r.into_inner();
202        file.seek(SeekFrom::End(0))?;
203
204        let file = PersistFile {
205            file: BufWriter::new(file),
206            needs_resynch: Ok(()),
207            lock,
208        };
209
210        Ok(Self {
211            seen,
212            file: Some(file),
213            replay_log_type: PhantomData,
214        })
215    }
216
217    /// Truncate `file` to contain a whole number of records
218    ///
219    /// `current_len` should have come from `file.metadata()`.
220    // If the file's length is not an even multiple of MESSAGE_LEN after the MAGIC, truncate it.
221    fn truncate_to_multiple(file: &mut File, current_len: u64) -> io::Result<()> {
222        let excess = (current_len - T::MAGIC.len() as u64) % (OUTPUT_LEN as u64);
223        if excess != 0 {
224            file.set_len(current_len - excess)?;
225        }
226        Ok(())
227    }
228
229    /// Test whether we have already seen `message`.
230    ///
231    /// If we have seen it, return `Err(ReplayError::AlreadySeen)`.  (Since this
232    /// is a probabilistic data structure, there is a chance of returning this
233    /// error even if we have we have _not_ seen this particular message)
234    ///
235    /// Otherwise, return `Ok(())`.
236    pub(crate) fn check_for_replay(&mut self, message: &T::Message) -> Result<(), ReplayError> {
237        let h = T::transform_message(message);
238        self.seen.test_and_add(&h)?;
239        if let Some(f) = self.file.as_mut() {
240            (|| {
241                // If write_all fails, it might have written part of the data;
242                // in that case, we must truncate the file to resynchronise.
243                // We set a note to truncate just before we call write_all
244                // and clear it again afterwards.
245                //
246                // But, first, we need to deal with any previous note we left ourselves.
247
248                // (With the current implementation of std::io::BufWriter, this is
249                // unnecessary, because if the argument to write_all is smaller than
250                // the buffer size, BufWriter::write_all always just copies to the buffer,
251                // flushing first if necessary; and when it flushes, it uses write,
252                // not write_all.  So the use of write_all never causes "lost" data.
253                // However, this is not a documented guarantee.)
254                match f.needs_resynch {
255                    Ok(()) => {}
256                    Err(()) => {
257                        // We're going to reach behind the BufWriter, so we need to make
258                        // sure it's in synch with the underlying File.
259                        f.file.flush()?;
260                        let inner = f.file.get_mut();
261                        let len = inner.metadata()?.len();
262                        Self::truncate_to_multiple(inner, len)?;
263                        // cursor is now past end, must reset (see std::fs::File::set_len)
264                        inner.seek(SeekFrom::End(0))?;
265                    }
266                }
267                f.needs_resynch = Err(());
268
269                f.file.write_all(&h[..])?;
270
271                f.needs_resynch = Ok(());
272
273                Ok(())
274            })()
275            .map_err(|e| ReplayError::Log(Arc::new(e)))?;
276        }
277        Ok(())
278    }
279
280    /// Flush any buffered data to disk.
281    #[allow(dead_code)] // TODO #1208
282    pub(crate) fn flush(&mut self) -> Result<(), io::Error> {
283        if let Some(f) = self.file.as_mut() {
284            f.file.flush()?;
285        }
286        Ok(())
287    }
288
289    /// Tries to parse a filename in the replay logs directory
290    ///
291    /// If the leafname refers to a file that would be created by
292    /// [`ReplayLog::new_logged`], returns the name as a Rust type.
293    ///
294    /// Otherwise returns an error explaining why it isn't,
295    /// as a plain string (for logging).
296    pub(crate) fn parse_log_leafname(leaf: &OsStr) -> Result<T::Name, Cow<'static, str>> {
297        T::parse_log_leafname(leaf)
298    }
299}
300
301/// Wrapper around a fast-ish data structure for detecting replays with some
302/// false positive rate.  Bloom filters, cuckoo filters, and xorf filters are all
303/// an option here.  You could even use a HashSet.
304///
305/// We isolate this code to make it easier to replace.
306mod data {
307    use super::{ReplayError, OUTPUT_LEN};
308    use growable_bloom_filter::GrowableBloom;
309
310    /// A probabilistic membership filter.
311    pub(super) struct Filter(pub(crate) GrowableBloom);
312
313    impl Filter {
314        /// Create a new empty filter
315        pub(super) fn new() -> Self {
316            // TODO: Perhaps we should make the capacity here tunable, based on
317            // the number of entries we expect.  These values are more or less
318            // pulled out of thin air.
319            let desired_error_prob = 1.0 / 100_000.0;
320            let est_insertions = 100_000;
321            Filter(GrowableBloom::new(desired_error_prob, est_insertions))
322        }
323
324        /// Try to add `msg` to this filter if it isn't already there.
325        ///
326        /// Return Ok(()) or Err(AlreadySeen).
327        pub(super) fn test_and_add(&mut self, msg: &[u8; OUTPUT_LEN]) -> Result<(), ReplayError> {
328            if self.0.insert(&msg[..]) {
329                Ok(())
330            } else {
331                Err(ReplayError::AlreadySeen)
332            }
333        }
334    }
335}
336
337/// A problem that prevents us from reading a ReplayLog from disk.
338///
339/// (This only exists so we can wrap it up in an [`io::Error`])
340#[derive(thiserror::Error, Clone, Debug)]
341enum LogContentError {
342    /// The magic number on the log file was incorrect.
343    #[error("unrecognized data format")]
344    UnrecognizedFormat,
345}
346
347/// An error occurred while checking whether we've seen an element before.
348#[derive(thiserror::Error, Clone, Debug)]
349pub(crate) enum ReplayError {
350    /// We have already seen this item.
351    #[error("Already seen")]
352    AlreadySeen,
353
354    /// We were unable to record this item in the log.
355    #[error("Unable to log data")]
356    Log(Arc<std::io::Error>),
357}
358
359#[cfg(test)]
360mod test {
361    // @@ begin test lint list maintained by maint/add_warning @@
362    #![allow(clippy::bool_assert_comparison)]
363    #![allow(clippy::clone_on_copy)]
364    #![allow(clippy::dbg_macro)]
365    #![allow(clippy::mixed_attributes_style)]
366    #![allow(clippy::print_stderr)]
367    #![allow(clippy::print_stdout)]
368    #![allow(clippy::single_char_pattern)]
369    #![allow(clippy::unwrap_used)]
370    #![allow(clippy::unchecked_duration_subtraction)]
371    #![allow(clippy::useless_vec)]
372    #![allow(clippy::needless_pass_by_value)]
373    //! <!-- @@ end test lint list maintained by maint/add_warning @@ -->
374
375    use super::*;
376    use crate::test::mk_state_instance;
377    use rand::Rng;
378    use test_temp_dir::{test_temp_dir, TestTempDir, TestTempDirGuard};
379
380    struct TestReplayLogType;
381
382    type TestReplayLog = ReplayLog<TestReplayLogType>;
383
384    impl ReplayLogType for TestReplayLogType {
385        type Name = IptLocalId;
386        type Message = [u8; OUTPUT_LEN];
387
388        const MAGIC: &'static [u8; MAGIC_LEN] = b"<tor test replay>\n\0\0\0\0\0\0\0\0\0\0\0\0\0\0";
389
390        fn format_filename(name: &IptLocalId) -> String {
391            format!("{name}{REPLAY_LOG_SUFFIX}")
392        }
393
394        fn transform_message(message: &[u8; OUTPUT_LEN]) -> [u8; OUTPUT_LEN] {
395            message.clone()
396        }
397
398        fn parse_log_leafname(leaf: &OsStr) -> Result<IptLocalId, Cow<'static, str>> {
399            let leaf = leaf.to_str().ok_or("not proper unicode")?;
400            let lid = leaf.strip_suffix(REPLAY_LOG_SUFFIX).ok_or("not *.bin")?;
401            let lid: IptLocalId = lid
402                .parse()
403                .map_err(|e: crate::InvalidIptLocalId| e.to_string())?;
404            Ok(lid)
405        }
406    }
407
408    fn rand_msg<R: Rng>(rng: &mut R) -> [u8; OUTPUT_LEN] {
409        rng.random()
410    }
411
412    /// Basic tests on an ephemeral IptReplayLog.
413    #[test]
414    fn simple_usage() {
415        let mut rng = tor_basic_utils::test_rng::testing_rng();
416        let group_1: Vec<_> = (0..=100).map(|_| rand_msg(&mut rng)).collect();
417        let group_2: Vec<_> = (0..=100).map(|_| rand_msg(&mut rng)).collect();
418
419        let mut log = TestReplayLog::new_ephemeral();
420        // Add everything in group 1.
421        for msg in &group_1 {
422            assert!(log.check_for_replay(msg).is_ok(), "False positive");
423        }
424        // Make sure that everything in group 1 is still there.
425        for msg in &group_1 {
426            assert!(log.check_for_replay(msg).is_err());
427        }
428        // Make sure that group 2 is detected as not-there.
429        for msg in &group_2 {
430            assert!(log.check_for_replay(msg).is_ok(), "False positive");
431        }
432    }
433
434    const TEST_TEMP_SUBDIR: &str = "replaylog";
435
436    fn create_logged(dir: &TestTempDir) -> TestTempDirGuard<TestReplayLog> {
437        dir.subdir_used_by(TEST_TEMP_SUBDIR, |dir| {
438            let inst = mk_state_instance(&dir, "allium");
439            let raw = inst.raw_subdir("iptreplay").unwrap();
440            TestReplayLog::new_logged(&raw, &IptLocalId::dummy(1)).unwrap()
441        })
442    }
443
444    /// Basic tests on an persistent IptReplayLog.
445    #[test]
446    fn logging_basics() {
447        let mut rng = tor_basic_utils::test_rng::testing_rng();
448        let group_1: Vec<_> = (0..=100).map(|_| rand_msg(&mut rng)).collect();
449        let group_2: Vec<_> = (0..=100).map(|_| rand_msg(&mut rng)).collect();
450
451        let dir = test_temp_dir!();
452        let mut log = create_logged(&dir);
453        // Add everything in group 1, then close and reload.
454        for msg in &group_1 {
455            assert!(log.check_for_replay(msg).is_ok(), "False positive");
456        }
457        drop(log);
458        let mut log = create_logged(&dir);
459        // Make sure everything in group 1 is still there.
460        for msg in &group_1 {
461            assert!(log.check_for_replay(msg).is_err());
462        }
463        // Now add everything in group 2, then close and reload.
464        for msg in &group_2 {
465            assert!(log.check_for_replay(msg).is_ok(), "False positive");
466        }
467        drop(log);
468        let mut log = create_logged(&dir);
469        // Make sure that groups 1 and 2 are still there.
470        for msg in group_1.iter().chain(group_2.iter()) {
471            assert!(log.check_for_replay(msg).is_err());
472        }
473    }
474
475    /// Test for a log that gets truncated mid-write.
476    #[test]
477    fn test_truncated() {
478        let mut rng = tor_basic_utils::test_rng::testing_rng();
479        let group_1: Vec<_> = (0..=100).map(|_| rand_msg(&mut rng)).collect();
480        let group_2: Vec<_> = (0..=100).map(|_| rand_msg(&mut rng)).collect();
481
482        let dir = test_temp_dir!();
483        let mut log = create_logged(&dir);
484        for msg in &group_1 {
485            assert!(log.check_for_replay(msg).is_ok(), "False positive");
486        }
487        drop(log);
488        // Truncate the file by 7 bytes.
489        dir.subdir_used_by(TEST_TEMP_SUBDIR, |dir| {
490            let path = dir.join(format!("hss/allium/iptreplay/{}.bin", IptLocalId::dummy(1)));
491            let file = OpenOptions::new().write(true).open(path).unwrap();
492            // Make sure that the file has the length we expect.
493            let expected_len = MAGIC_LEN + OUTPUT_LEN * group_1.len();
494            assert_eq!(expected_len as u64, file.metadata().unwrap().len());
495            file.set_len((expected_len - 7) as u64).unwrap();
496        });
497        // Now, reload the log. We should be able to recover every non-truncated
498        // item...
499        let mut log = create_logged(&dir);
500        for msg in &group_1[..group_1.len() - 1] {
501            assert!(log.check_for_replay(msg).is_err());
502        }
503        // But not the last one, which we truncated.  (Checking will add it, though.)
504        assert!(
505            log.check_for_replay(&group_1[group_1.len() - 1]).is_ok(),
506            "False positive"
507        );
508        // Now add everything in group 2, then close and reload.
509        for msg in &group_2 {
510            assert!(log.check_for_replay(msg).is_ok(), "False positive");
511        }
512        drop(log);
513        let mut log = create_logged(&dir);
514        // Make sure that groups 1 and 2 are still there.
515        for msg in group_1.iter().chain(group_2.iter()) {
516            assert!(log.check_for_replay(msg).is_err());
517        }
518    }
519
520    /// Test for a partial write
521    #[test]
522    #[cfg(target_os = "linux")] // different platforms have different definitions of sigaction
523    fn test_partial_write() {
524        use std::env;
525        use std::os::unix::process::ExitStatusExt;
526        use std::process::Command;
527
528        // TODO this contraption should perhaps be productised and put somewhere else
529
530        const ENV_NAME: &str = "TOR_HSSERVICE_TEST_PARTIAL_WRITE_SUBPROCESS";
531        // for a wait status different from any of libtest's
532        const GOOD_SIGNAL: i32 = libc::SIGUSR2;
533
534        let sigemptyset = || unsafe {
535            let mut set = MaybeUninit::uninit();
536            libc::sigemptyset(set.as_mut_ptr());
537            set.assume_init()
538        };
539
540        // Check that SIGUSR2 starts out as SIG_DFL and unblocked
541        //
542        // We *reject* such situations, rather than fixing them up, because this is an
543        // irregular and broken environment that can cause arbitrarily weird behaviours.
544        // Programs on Unix are entitled to assume that their signal dispositions are
545        // SIG_DFL on entry, with signals unblocked.  (With a few exceptions.)
546        //
547        // So we want to detect and report any such environment, not let it slide.
548        unsafe {
549            let mut sa = MaybeUninit::uninit();
550            let r = libc::sigaction(GOOD_SIGNAL, ptr::null(), sa.as_mut_ptr());
551            assert_eq!(r, 0);
552            let sa = sa.assume_init();
553            assert_eq!(
554                sa.sa_sigaction,
555                libc::SIG_DFL,
556                "tests running in broken environment (SIGUSR2 not SIG_DFL)"
557            );
558
559            let empty_set = sigemptyset();
560            let mut current_set = MaybeUninit::uninit();
561            let r = libc::sigprocmask(
562                libc::SIG_UNBLOCK,
563                (&empty_set) as _,
564                current_set.as_mut_ptr(),
565            );
566            assert_eq!(r, 0);
567            let current_set = current_set.assume_init();
568            let blocked = libc::sigismember((&current_set) as _, GOOD_SIGNAL);
569            assert_eq!(
570                blocked, 0,
571                "tests running in broken environment (SIGUSR2 blocked)"
572            );
573        }
574
575        match env::var(ENV_NAME) {
576            Err(env::VarError::NotPresent) => {
577                eprintln!("in test runner process, forking..,");
578                let output = Command::new(env::current_exe().unwrap())
579                    .args(["--nocapture", "replay::test::test_partial_write"])
580                    .env(ENV_NAME, "1")
581                    .output()
582                    .unwrap();
583                let print_output = |prefix, data| match std::str::from_utf8(data) {
584                    Ok(s) => {
585                        for l in s.split("\n") {
586                            eprintln!(" {prefix} {l}");
587                        }
588                    }
589                    Err(e) => eprintln!(" UTF-8 ERROR {prefix} {e}"),
590                };
591                print_output("!", &output.stdout);
592                print_output(">", &output.stderr);
593                let st = output.status;
594                eprintln!("reaped actual test process {st:?} (expecting signal {GOOD_SIGNAL})");
595                assert_eq!(st.signal(), Some(GOOD_SIGNAL));
596                return;
597            }
598            Ok(y) if y == "1" => {}
599            other => panic!("bad env var {ENV_NAME:?} {other:?}"),
600        };
601
602        // Now we are in our own process, and can mess about with ulimit etc.
603
604        use std::fs;
605        use std::mem::MaybeUninit;
606        use std::ptr;
607
608        fn set_ulimit(size: usize) {
609            unsafe {
610                use libc::RLIMIT_FSIZE;
611                let mut rlim = libc::rlimit {
612                    rlim_cur: 0,
613                    rlim_max: 0,
614                };
615                let r = libc::getrlimit(RLIMIT_FSIZE, (&mut rlim) as _);
616                assert_eq!(r, 0);
617                rlim.rlim_cur = size.try_into().unwrap();
618                let r = libc::setrlimit(RLIMIT_FSIZE, (&rlim) as _);
619                assert_eq!(r, 0);
620            }
621        }
622
623        // This test is quite complicated.
624        //
625        // We want to test partial writes.  We could perhaps have done this by
626        // parameterising IptReplayLog so it could have something other than File,
627        // but that would probably leak into the public API.
628        //
629        // Instead, we cause *actual* partial writes.  We use the Unix setrlimit
630        // call to limit the size of files our process is allowed to write.
631        // This causes the underlying write(2) calls to (i) generate SIGXFSZ
632        // (ii) if that doesn't kill the process, return partial writes.
633
634        test_temp_dir!().used_by(|dir| {
635            let path = dir.join("test.log");
636            let lock = LockFileGuard::lock(dir.join("dummy.lock")).unwrap();
637            let lock = Arc::new(lock);
638            let mut rl = TestReplayLog::new_logged_inner(&path, lock.clone()).unwrap();
639
640            const BUF: usize = 8192; // BufWriter default; if that changes, test will break
641
642            // We let ourselves write one whole buffer plus an odd amount of extra
643            const ALLOW: usize = BUF + 37;
644
645            // Ignore SIGXFSZ (default disposition is for exceeding the rlimit to kill us)
646            unsafe {
647                let sa = libc::sigaction {
648                    sa_sigaction: libc::SIG_IGN,
649                    sa_mask: sigemptyset(),
650                    sa_flags: 0,
651                    sa_restorer: None,
652                };
653                let r = libc::sigaction(libc::SIGXFSZ, (&sa) as _, ptr::null_mut());
654                assert_eq!(r, 0);
655            }
656
657            let demand_efbig = |e| match e {
658                // TODO MSRV 1.83: replace with io::ErrorKind::FileTooLarge?
659                ReplayError::Log(e) if e.raw_os_error() == Some(libc::EFBIG) => {}
660                other => panic!("expected EFBUG, got {other:?}"),
661            };
662
663            // Generate a distinct message given a phase and a counter
664            #[allow(clippy::identity_op)]
665            let mk_msg = |phase: u8, i: usize| {
666                let i = u32::try_from(i).unwrap();
667                let mut msg = [0_u8; OUTPUT_LEN];
668                msg[0] = phase;
669                msg[1] = phase;
670                msg[4] = (i >> 24) as _;
671                msg[5] = (i >> 16) as _;
672                msg[6] = (i >> 8) as _;
673                msg[7] = (i >> 0) as _;
674                msg
675            };
676
677            // Number of hashes we can write to the file before failure occurs
678            const CAN_DO: usize = (ALLOW + BUF - MAGIC_LEN) / OUTPUT_LEN;
679            dbg!(MAGIC_LEN, OUTPUT_LEN, BUF, ALLOW, CAN_DO);
680
681            // Record of the hashes that TestReplayLog tells us were OK and not replays;
682            // ie, which it therefore ought to have recorded.
683            let mut gave_ok = Vec::new();
684
685            set_ulimit(ALLOW);
686
687            for i in 0..CAN_DO {
688                let h = mk_msg(b'y', i);
689                rl.check_for_replay(&h).unwrap();
690                gave_ok.push(h);
691            }
692
693            let md = fs::metadata(&path).unwrap();
694            dbg!(md.len(), &rl.file);
695
696            // Now we have written what we can.  The next two calls will fail,
697            // since the BufWriter buffer is full and can't be flushed.
698
699            for i in 0..2 {
700                eprintln!("expecting EFBIG {i}");
701                demand_efbig(rl.check_for_replay(&mk_msg(b'n', i)).unwrap_err());
702                let md = fs::metadata(&path).unwrap();
703                assert_eq!(md.len(), u64::try_from(ALLOW).unwrap());
704            }
705
706            // Enough that we don't get any further file size exceedances
707            set_ulimit(ALLOW * 10);
708
709            // Now we should be able to recover.  We write two more hashes.
710            for i in 0..2 {
711                eprintln!("recovering {i}");
712                let h = mk_msg(b'r', i);
713                rl.check_for_replay(&h).unwrap();
714                gave_ok.push(h);
715            }
716
717            // flush explicitly just so we catch any error
718            // (drop would flush, but it can't report errors)
719            rl.flush().unwrap();
720            drop(rl);
721
722            // Reopen the log - reading in the written data.
723            // We can then check that everything the earlier IptReplayLog
724            // claimed to have written, is indeed recorded.
725
726            let mut rl = TestReplayLog::new_logged_inner(&path, lock.clone()).unwrap();
727            for msg in &gave_ok {
728                match rl.check_for_replay(msg) {
729                    Err(ReplayError::AlreadySeen) => {}
730                    other => panic!("expected AlreadySeen, got {other:?}"),
731                }
732            }
733
734            eprintln!("recovered file contents checked, all good");
735        });
736
737        unsafe {
738            libc::raise(libc::SIGUSR2);
739        }
740        panic!("we survived raise SIGUSR2");
741    }
742}