kumo_server_common/
acct.rs

1use crate::authn_authz::{Access, AuditRecord, AuthInfo, Identity};
2use crate::disk_space::{MinFree, MonitoredPath};
3use crate::log::{mark_existing_logs_as_done_in_dir, OpenedFile};
4use anyhow::Context;
5use chrono::{DateTime, Utc};
6use parking_lot::Mutex;
7use serde::{Deserialize, Serialize};
8use std::io::Write;
9use std::path::PathBuf;
10use std::sync::OnceLock;
11use std::time::{Duration, Instant};
12use tokio::sync::mpsc::{channel, Receiver, Sender};
13use tokio::task::JoinHandle;
14use zstd::Encoder;
15
16static LOGGER: OnceLock<AcctLogger> = OnceLock::new();
17
18#[derive(Debug)]
19enum AcctCommand {
20    Record(AcctLogRecord),
21    /// Stop after writing every record queued ahead of this command.
22    Terminate,
23}
24
25struct AcctLogger {
26    tx: Sender<AcctCommand>,
27    /// Present until `shutdown` takes it to join the logger task.
28    task: Mutex<Option<JoinHandle<()>>>,
29}
30
31#[derive(Deserialize, Clone, Debug)]
32#[serde(deny_unknown_fields)]
33pub struct AuditLogParams {
34    /// Where to place the log files
35    pub log_dir: PathBuf,
36    /// How many uncompressed bytes to allow per file segment
37    #[serde(default = "AuditLogParams::default_max_file_size")]
38    pub max_file_size: u64,
39    /// Maximum number of outstanding items to be logged before
40    /// the submission will block; helps to avoid runaway issues
41    /// spiralling out of control.
42    #[serde(default = "AuditLogParams::default_back_pressure")]
43    pub back_pressure: usize,
44
45    /// The level of compression.
46    /// 0 - use the zstd default level (probably 3).
47    /// 1-21 are the explicitly configurable levels
48    #[serde(default = "AuditLogParams::default_compression_level")]
49    pub compression_level: i32,
50
51    #[serde(default, with = "duration_serde")]
52    pub max_segment_duration: Option<Duration>,
53
54    #[serde(default)]
55    pub min_free_space: MinFree,
56    #[serde(default)]
57    pub min_free_inodes: MinFree,
58
59    /// Log records for successfully granted authz
60    #[serde(default = "AuditLogParams::default_true")]
61    pub log_authz_allow: bool,
62
63    /// Log records for denied authz
64    #[serde(default = "AuditLogParams::default_true")]
65    pub log_authz_deny: bool,
66
67    /// Log records for successful authn
68    #[serde(default = "AuditLogParams::default_true")]
69    pub log_authn_ok: bool,
70
71    /// Log records for failed authn
72    #[serde(default = "AuditLogParams::default_true")]
73    pub log_authn_fail: bool,
74}
75
76impl AuditLogParams {
77    pub fn default_max_file_size() -> u64 {
78        1_000_000_000
79    }
80    pub fn default_back_pressure() -> usize {
81        128_000
82    }
83    pub fn default_compression_level() -> i32 {
84        0 // use the zstd default
85    }
86
87    pub fn default_true() -> bool {
88        true
89    }
90
91    pub async fn init(&self) -> anyhow::Result<()> {
92        let (tx, rx) = channel(self.back_pressure);
93        let params = self.clone();
94        let task = kumo_server_runtime::spawn_on_main("acct logger", async move {
95            if let Err(err) = params.acct_logger(rx).await {
96                tracing::error!(
97                    "AuditLogParams::acct_logger returned with error: {err}, \
98                    audit logging will cease to function until the next restart!"
99                );
100            }
101        })?;
102
103        let logger = AcctLogger {
104            tx,
105            task: Mutex::new(Some(task)),
106        };
107        if LOGGER.set(logger).is_err() {
108            anyhow::bail!("AuditLogParams::init must not be called more than once");
109        }
110
111        Ok(())
112    }
113
114    async fn acct_logger(self, mut rx: Receiver<AcctCommand>) -> anyhow::Result<()> {
115        if !config::is_validating() {
116            if let Err(err) = mark_existing_logs_as_done_in_dir(&self.log_dir) {
117                tracing::error!("{err}");
118            }
119            std::fs::create_dir_all(&self.log_dir).ok();
120            MonitoredPath {
121                name: format!("log dir {}", self.log_dir.display()),
122                path: self.log_dir.clone(),
123                min_free_space: self.min_free_space,
124                min_free_inodes: self.min_free_inodes,
125            }
126            .register();
127        }
128
129        let mut expire_counter = 0u16;
130        let mut current_file: Option<OpenedFile> = None;
131
132        loop {
133            let deadline = current_file.as_ref().and_then(|f| f.expires);
134
135            let mut check_expire = false;
136            let record = match deadline {
137                Some(deadline) => tokio::select! {
138                    cmd = rx.recv() => {
139                        match cmd {
140                            Some(AcctCommand::Record(r)) => Some(r),
141                            Some(AcctCommand::Terminate) | None => return Ok(()),
142                        }
143                    }
144                    _ = tokio::time::sleep_until(deadline.into()) => {
145                        check_expire = true;
146                        None
147                    }
148                },
149                None => match rx.recv().await {
150                    Some(AcctCommand::Record(r)) => Some(r),
151                    Some(AcctCommand::Terminate) | None => return Ok(()),
152                },
153            };
154
155            expire_counter = expire_counter.wrapping_add(1);
156            if expire_counter == 10_000 {
157                check_expire = true;
158            }
159
160            if check_expire {
161                let now = Instant::now();
162                let do_expire = current_file
163                    .as_ref()
164                    .and_then(|f| f.expires.map(|exp| exp <= now))
165                    .unwrap_or(false);
166
167                if do_expire {
168                    current_file.take();
169                }
170                expire_counter = 0;
171            }
172
173            if let Some(record) = record {
174                let should_log = match &record {
175                    AcctLogRecord::Authentication(authn) => {
176                        if authn.success {
177                            self.log_authn_ok
178                        } else {
179                            self.log_authn_fail
180                        }
181                    }
182                    AcctLogRecord::Authorization(authz) => match authz.access {
183                        Access::Allow => self.log_authz_allow,
184                        Access::Deny => self.log_authz_deny,
185                    },
186                };
187
188                if !should_log {
189                    continue;
190                }
191
192                if let Err(err) = self.open_file_if_needed(&mut current_file) {
193                    tracing::error!("{err}");
194                    continue;
195                }
196
197                if let Err(err) = self.do_record(&mut current_file, record) {
198                    tracing::error!("{err}");
199                    continue;
200                }
201            }
202        }
203    }
204
205    fn open_file_if_needed(&self, current_file: &mut Option<OpenedFile>) -> anyhow::Result<()> {
206        if current_file.is_some() {
207            return Ok(());
208        }
209        let now = Utc::now();
210
211        let base_name = now.format("acct-%Y%m%d-%H%M%S%.f").to_string();
212
213        let name = self.log_dir.join(base_name);
214        // They might be trying to use multiple directories below
215        // the configured log_dir, so adjust our idea of its parent
216        // dir
217        let log_dir = name
218            .parent()
219            .expect("log_dir.join ensures we always have a parent");
220
221        let f = match std::fs::OpenOptions::new()
222            .append(true)
223            .create(true)
224            .open(&name)
225        {
226            Ok(f) => f,
227            Err(err) => {
228                if err.kind() == std::io::ErrorKind::NotFound {
229                    match std::fs::create_dir_all(log_dir) {
230                        Ok(_) => {
231                            // Try opening it again now
232                            std::fs::OpenOptions::new()
233                                .append(true)
234                                .create(true)
235                                .open(&name)
236                                .with_context(|| format!("open log file {name:?}"))?
237                        }
238                        Err(dir_err) => {
239                            anyhow::bail!(
240                                "open log file {name:?}: failed: {err:#?}. \
241                                    Additionally, attempting to create dir {} failed: {dir_err:#?}",
242                                log_dir.display()
243                            );
244                        }
245                    }
246                } else {
247                    anyhow::bail!("open log file {name:?}: failed: {err:#?}");
248                }
249            }
250        };
251
252        let file = OpenedFile {
253            file: Encoder::new(f, self.compression_level).context("set up zstd encoder")?,
254            name,
255            written: 0,
256            expires: self
257                .max_segment_duration
258                .map(|duration| Instant::now() + duration),
259        };
260
261        current_file.replace(file);
262        Ok(())
263    }
264
265    fn do_record(
266        &self,
267        current_file: &mut Option<OpenedFile>,
268        record: AcctLogRecord,
269    ) -> anyhow::Result<()> {
270        if let Some(file) = current_file {
271            let mut record_text = Vec::new();
272            serde_json::to_writer(&mut record_text, &record).context("serializing record")?;
273            record_text.push(b'\n');
274
275            file.file
276                .write_all(&record_text)
277                .with_context(|| format!("writing record to {}", file.name.display()))?;
278            file.written += record_text.len() as u64;
279
280            let need_rotate = file.written >= self.max_file_size
281                || file
282                    .expires
283                    .map(|exp| exp <= Instant::now())
284                    .unwrap_or(false);
285
286            if need_rotate {
287                current_file.take();
288            }
289        }
290        Ok(())
291    }
292}
293
294#[derive(Serialize, Deserialize, Debug)]
295#[serde(tag = "type")]
296pub enum AcctLogRecord {
297    Authentication(AuthnAuditRecord),
298    Authorization(AuditRecord),
299}
300
301impl AcctLogRecord {
302    pub fn timestamp(&self) -> &DateTime<Utc> {
303        match self {
304            Self::Authentication(authn) => &authn.timestamp,
305            Self::Authorization(authz) => &authz.timestamp,
306        }
307    }
308
309    pub fn is_allow(&self) -> bool {
310        match self {
311            Self::Authentication(authn) => authn.success,
312            Self::Authorization(authz) => authz.access == Access::Allow,
313        }
314    }
315}
316
317async fn log_acct(record: AcctLogRecord) -> anyhow::Result<()> {
318    tracing::trace!("Audit: {record:?}");
319    if let Some(logger) = LOGGER.get() {
320        logger.tx.send(AcctCommand::Record(record)).await?;
321    }
322    Ok(())
323}
324
325/// Stop the audit logger and finalize its open log segment. Records
326/// submitted before this call are written first. Blocks until the
327/// logger task has exited. Does nothing if the logger was never
328/// configured.
329pub async fn shutdown() {
330    let Some(logger) = LOGGER.get() else {
331        return;
332    };
333    logger.tx.send(AcctCommand::Terminate).await.ok();
334    let task = logger.task.lock().take();
335    if let Some(task) = task {
336        if let Err(err) = task.await {
337            tracing::error!("acct logger task failed: {err:#}");
338        }
339    }
340}
341
342pub async fn log_authz(record: AuditRecord) -> anyhow::Result<()> {
343    log_acct(AcctLogRecord::Authorization(record)).await
344}
345
346#[derive(Serialize, Deserialize, Debug)]
347pub struct AuthnAuditRecord {
348    /// When the event occurred
349    pub timestamp: DateTime<Utc>,
350    /// Who they were attempting to authenticate as
351    pub attempted_identity: Identity,
352    /// Whether the attempt was successful
353    pub success: bool,
354    /// The resulting authentication info
355    pub auth_info: AuthInfo,
356}
357
358pub async fn log_authn(record: AuthnAuditRecord) -> anyhow::Result<()> {
359    log_acct(AcctLogRecord::Authentication(record)).await
360}
361
362#[cfg(test)]
363mod test {
364    use super::*;
365    use crate::authn_authz::IdentityContext;
366
367    /// Builds params that log everything into `log_dir`, with no time based
368    /// rotation and the given size limit.
369    fn params(log_dir: &std::path::Path, max_file_size: u64) -> AuditLogParams {
370        AuditLogParams {
371            log_dir: log_dir.to_path_buf(),
372            max_file_size,
373            back_pressure: 16,
374            compression_level: AuditLogParams::default_compression_level(),
375            max_segment_duration: None,
376            min_free_space: MinFree::default(),
377            min_free_inodes: MinFree::default(),
378            log_authz_allow: true,
379            log_authz_deny: true,
380            log_authn_ok: true,
381            log_authn_fail: true,
382        }
383    }
384
385    fn authn_record(identity: &str) -> AcctLogRecord {
386        AcctLogRecord::Authentication(AuthnAuditRecord {
387            timestamp: Utc::now(),
388            attempted_identity: Identity {
389                identity: identity.to_string(),
390                context: IdentityContext::SmtpAuthPlainAuthentication,
391            },
392            success: true,
393            auth_info: AuthInfo::default(),
394        })
395    }
396
397    /// Decode every segment in `dir`, returning the identity from each
398    /// record in filename order.
399    fn read_identities(dir: &std::path::Path) -> anyhow::Result<Vec<Vec<String>>> {
400        let mut paths: Vec<_> = std::fs::read_dir(dir)?
401            .map(|entry| entry.map(|e| e.path()))
402            .collect::<Result<_, _>>()?;
403        paths.sort();
404        paths
405            .iter()
406            .map(|path| {
407                let data = zstd::stream::decode_all(std::fs::File::open(path)?)?;
408                String::from_utf8(data)?
409                    .lines()
410                    .map(|line| match serde_json::from_str(line)? {
411                        AcctLogRecord::Authentication(authn) => {
412                            Ok(authn.attempted_identity.identity)
413                        }
414                        other => anyhow::bail!("unexpected record {other:?}"),
415                    })
416                    .collect()
417            })
418            .collect()
419    }
420
421    /// With no size or time based rotation, shutdown is required to complete
422    /// the zstd frame.
423    #[tokio::test]
424    async fn shutdown_finalizes_open_segment() -> anyhow::Result<()> {
425        let dir = tempfile::tempdir()?;
426        params(dir.path(), AuditLogParams::default_max_file_size())
427            .init()
428            .await?;
429
430        let AcctLogRecord::Authentication(record) = authn_record("daniel") else {
431            unreachable!()
432        };
433        log_authn(record).await?;
434
435        shutdown().await;
436
437        k9::assert_equal!(
438            read_identities(dir.path())?,
439            vec![vec!["daniel".to_string()]]
440        );
441        Ok(())
442    }
443
444    #[test]
445    fn size_limit_rotates_segment() -> anyhow::Result<()> {
446        let dir = tempfile::tempdir()?;
447        let params = params(dir.path(), 1);
448        let mut current_file = None;
449
450        for identity in ["first", "second"] {
451            params.open_file_if_needed(&mut current_file)?;
452            params.do_record(&mut current_file, authn_record(identity))?;
453            assert!(
454                current_file.is_none(),
455                "segment must rotate after {identity}"
456            );
457        }
458
459        k9::assert_equal!(
460            read_identities(dir.path())?,
461            vec![vec!["first".to_string()], vec!["second".to_string()]]
462        );
463        Ok(())
464    }
465}
kumo_server_common/acct.rs

kumo_server_common/
acct.rs
