elasticctl 0.7.4

Operate Elastic Security as code with a safety-first CLI for security engineers.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99

use super::filesystem::*;
use super::{FileOps, Journal, Phase, PhaseRecord, PreparedRecord, with_newline};
use elasticctl_core::{Error, ErrorKind, Result};
use std::fs;
use std::path::Path;

pub(super) fn append_phase(
    transaction_dir: &Path,
    phase: Phase,
    target: &Path,
    ops: &dyn FileOps,
) -> Result<()> {
    let bytes = serde_json::to_vec(&PhaseRecord { phase })
        .map_err(|e| Error::new(ErrorKind::Error, format!("encoding report journal: {e}")))?;
    let journal = journal_path(transaction_dir);
    write(
        ops,
        &journal,
        &with_newline(bytes),
        true,
        target,
        "appending report journal",
    )?;
    sync_file(ops, &journal, target, "syncing report journal")?;
    sync_dir(ops, transaction_dir, "syncing report transaction directory")
}

pub(super) fn read_journal(transaction_dir: &Path) -> Result<Option<Journal>> {
    let journal = journal_path(transaction_dir);
    if !path_exists(&journal, "checking report journal")? {
        return Ok(None);
    }
    require_regular_file(&journal, transaction_dir, "reading report journal")?;
    let bytes =
        fs::read(&journal).map_err(|e| filesystem_error("reading report journal", &journal, e))?;
    let mut records = bytes.split_inclusive(|byte| *byte == b'\n');
    let Some(first) = records.next() else {
        return Ok(None);
    };
    if !first.ends_with(b"\n") {
        return Ok(None);
    }
    let prepared: PreparedRecord = serde_json::from_slice(first).map_err(|e| {
        Error::new(
            ErrorKind::Error,
            format!("reading prepared report journal {}: {e}", journal.display()),
        )
    })?;
    if prepared.phase != Phase::Prepared {
        return Err(Error::new(
            ErrorKind::Error,
            format!(
                "report journal {} has a non-prepared first record",
                journal.display()
            ),
        ));
    }

    let mut phase = Phase::Prepared;
    for record in records {
        if !record.ends_with(b"\n") {
            // A crash can leave only the final append incomplete. It cannot
            // hide any later record because the incomplete slice is final.
            break;
        }
        let update: PhaseRecord = serde_json::from_slice(record).map_err(|e| {
            Error::new(
                ErrorKind::Error,
                format!("reading report journal {}: {e}", journal.display()),
            )
        })?;
        if !is_valid_transition(phase, update.phase) {
            return Err(Error::new(
                ErrorKind::Error,
                format!(
                    "report journal {} has an invalid phase transition",
                    journal.display()
                ),
            ));
        }
        phase = update.phase;
    }
    Ok(Some(Journal {
        existed_before: prepared.existed_before,
        phase,
    }))
}

fn is_valid_transition(from: Phase, to: Phase) -> bool {
    match from {
        Phase::Prepared => to == Phase::BackingUp,
        Phase::BackingUp => matches!(to, Phase::BackedUp | Phase::Restoring),
        Phase::BackedUp => matches!(to, Phase::Installing | Phase::Restoring),
        Phase::Installing => matches!(to, Phase::Installed | Phase::Restoring),
        Phase::Installed => to == Phase::Restoring,
        Phase::Restoring => to == Phase::Restored,
        Phase::Restored => false,
    }
}