v0.24.9

src/logging.rs

@@ -16,20 +16,23 @@ use serde::{Deserialize, Serialize};
 use serde_json::{json, Map, Value};
 use std::{
     collections::{BTreeMap, VecDeque},
+    io::{self, Write as IoWrite},
     path::PathBuf,
     sync::{
         atomic::{AtomicU64, Ordering},
-        Arc, Mutex,
+        Arc, Mutex, OnceLock,
     },
-    time::{Duration, SystemTime, UNIX_EPOCH},
+    time::{Duration, Instant, SystemTime, UNIX_EPOCH},
 };
 use tokio::{
     fs::{File, OpenOptions},
     io::AsyncWriteExt,
     net::TcpStream,
-    sync::{broadcast, mpsc},
+    sync::{broadcast, mpsc, watch},
 };
 use tokio_util::sync::CancellationToken;
+use tracing::{Event, Level, Subscriber};
+use tracing_subscriber::{layer::Context as LayerContext, Layer};
 
 pub const LIVE_WINDOW: usize = 256;
 pub const HISTORY_CAPACITY: usize = 4096;
@@ -238,6 +241,411 @@ fn source_value_name(name: &str) -> bool {
     )
 }
 
+
+#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
+pub enum ErrorLogFormat {
+    #[serde(rename = "JSON")]
+    Json,
+    #[serde(rename = "RAW")]
+    Raw,
+}
+impl Default for ErrorLogFormat {
+    fn default() -> Self { Self::Json }
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(default, deny_unknown_fields)]
+pub struct ErrorLogSettings {
+    pub enabled: bool,
+    pub path: String,
+    pub format: ErrorLogFormat,
+}
+impl Default for ErrorLogSettings {
+    fn default() -> Self {
+        Self { enabled: false, path: "/var/log/rustproxy/error.log".into(), format: ErrorLogFormat::Json }
+    }
+}
+impl ErrorLogSettings {
+    pub fn validate(&self) -> Result<()> {
+        if self.enabled {
+            ensure!(
+                !self.path.trim().is_empty() && self.path.len() <= 1024 && !self.path.chars().any(char::is_control),
+                "error_log.path must be 1..1024 bytes without controls"
+            );
+        }
+        Ok(())
+    }
+}
+
+const ERROR_LOG_QUEUE: usize = 512;
+const ERROR_LOG_RETRY: Duration = Duration::from_secs(30);
+
+#[derive(Debug, Clone, Serialize)]
+pub struct ErrorLogStats {
+    pub queue_capacity: usize,
+    pub queue_dropped: u64,
+    pub written: u64,
+    pub errors: u64,
+}
+
+#[derive(Debug)]
+struct ErrorEvent {
+    timestamp: String,
+    target: String,
+    message: String,
+    fields: BTreeMap<String, String>,
+}
+
+pub struct ErrorLogController {
+    settings: Arc<ArcSwap<ErrorLogSettings>>,
+    settings_changed: watch::Sender<()>,
+    sender: mpsc::Sender<ErrorEvent>,
+    dropped: Arc<AtomicU64>,
+    written: Arc<AtomicU64>,
+    errors: Arc<AtomicU64>,
+}
+impl ErrorLogController {
+    pub fn start() -> Arc<Self> {
+        let (sender, receiver) = mpsc::channel(ERROR_LOG_QUEUE);
+        let (settings_changed, settings_watch) = watch::channel(());
+        let settings = Arc::new(ArcSwap::from_pointee(ErrorLogSettings::default()));
+        let dropped = Arc::new(AtomicU64::new(0));
+        let written = Arc::new(AtomicU64::new(0));
+        let errors = Arc::new(AtomicU64::new(0));
+        tokio::spawn(error_log_loop(
+            receiver,
+            settings.clone(),
+            settings_watch,
+            written.clone(),
+            errors.clone(),
+        ));
+        Arc::new(Self { settings, settings_changed, sender, dropped, written, errors })
+    }
+    pub fn configure(&self, settings: ErrorLogSettings) {
+        self.settings.store(Arc::new(settings));
+        self.settings_changed.send_replace(());
+    }
+    fn push(&self, event: ErrorEvent) {
+        if self.settings.load().enabled && self.sender.try_send(event).is_err() {
+            self.dropped.fetch_add(1, Ordering::Relaxed);
+        }
+    }
+    pub fn stats(&self) -> ErrorLogStats {
+        ErrorLogStats {
+            queue_capacity: ERROR_LOG_QUEUE,
+            queue_dropped: self.dropped.load(Ordering::Relaxed),
+            written: self.written.load(Ordering::Relaxed),
+            errors: self.errors.load(Ordering::Relaxed),
+        }
+    }
+}
+
+static ERROR_LOG: OnceLock<Arc<ErrorLogController>> = OnceLock::new();
+
+pub fn install_error_log_controller(controller: Arc<ErrorLogController>) -> Result<()> {
+    ERROR_LOG.set(controller).map_err(|_| anyhow::anyhow!("error logger already initialized"))
+}
+pub fn configure_error_log(settings: ErrorLogSettings) {
+    if let Some(controller) = ERROR_LOG.get() { controller.configure(settings); }
+}
+pub fn error_log_stats() -> Option<ErrorLogStats> {
+    ERROR_LOG.get().map(|controller| controller.stats())
+}
+
+#[derive(Clone)]
+pub struct ErrorLogLayer {
+    controller: Arc<ErrorLogController>,
+}
+impl ErrorLogLayer {
+    pub fn new(controller: Arc<ErrorLogController>) -> Self { Self { controller } }
+}
+
+#[derive(Default)]
+struct EventVisitor {
+    message: Option<String>,
+    fields: BTreeMap<String, String>,
+}
+impl tracing::field::Visit for EventVisitor {
+    fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) {
+        let value = format!("{value:?}");
+        if field.name() == "message" { self.message = Some(value); }
+        else { self.fields.insert(field.name().to_owned(), value); }
+    }
+    fn record_str(&mut self, field: &tracing::field::Field, value: &str) {
+        if field.name() == "message" { self.message = Some(value.to_owned()); }
+        else { self.fields.insert(field.name().to_owned(), value.to_owned()); }
+    }
+    fn record_i64(&mut self, field: &tracing::field::Field, value: i64) { self.fields.insert(field.name().to_owned(), value.to_string()); }
+    fn record_u64(&mut self, field: &tracing::field::Field, value: u64) { self.fields.insert(field.name().to_owned(), value.to_string()); }
+    fn record_bool(&mut self, field: &tracing::field::Field, value: bool) { self.fields.insert(field.name().to_owned(), value.to_string()); }
+    fn record_f64(&mut self, field: &tracing::field::Field, value: f64) { self.fields.insert(field.name().to_owned(), value.to_string()); }
+}
+impl<S: Subscriber> Layer<S> for ErrorLogLayer {
+    fn on_event(&self, event: &Event<'_>, _ctx: LayerContext<'_, S>) {
+        if *event.metadata().level() != Level::ERROR { return; }
+        let mut visitor = EventVisitor::default();
+        event.record(&mut visitor);
+        let mut fields = visitor.fields;
+        for (key, value) in &mut fields {
+            *value = if is_secret_field(key) {
+                "[REDACTED]".into()
+            } else {
+                redact_text(value)
+            };
+        }
+        self.controller.push(ErrorEvent {
+            timestamp: rfc3339_now(),
+            target: event.metadata().target().to_owned(),
+            message: redact_text(visitor.message.as_deref().unwrap_or("error")),
+            fields,
+        });
+    }
+}
+
+#[derive(Clone, Copy, Default)]
+pub struct RedactingStderr;
+pub struct RedactingWriter { buffer: Vec<u8> }
+impl IoWrite for RedactingWriter {
+    fn write(&mut self, bytes: &[u8]) -> io::Result<usize> { self.buffer.extend_from_slice(bytes); Ok(bytes.len()) }
+    fn flush(&mut self) -> io::Result<()> {
+        if self.buffer.is_empty() { return Ok(()); }
+        let text = String::from_utf8_lossy(&self.buffer);
+        let redacted = redact_text(&text);
+        let mut stderr = io::stderr().lock();
+        stderr.write_all(redacted.as_bytes())?;
+        stderr.flush()?;
+        self.buffer.clear();
+        Ok(())
+    }
+}
+impl Drop for RedactingWriter { fn drop(&mut self) { let _ = self.flush(); } }
+impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for RedactingStderr {
+    type Writer = RedactingWriter;
+    fn make_writer(&'a self) -> Self::Writer { RedactingWriter { buffer: Vec::with_capacity(512) } }
+}
+
+fn redact_regex(pattern: &'static str) -> &'static regex::Regex {
+    static AUTH: OnceLock<regex::Regex> = OnceLock::new();
+    static QUERY: OnceLock<regex::Regex> = OnceLock::new();
+    static JSON: OnceLock<regex::Regex> = OnceLock::new();
+    static KV: OnceLock<regex::Regex> = OnceLock::new();
+    static PEM: OnceLock<regex::Regex> = OnceLock::new();
+    match pattern {
+        "auth" => AUTH.get_or_init(|| regex::Regex::new(r"(?i)(authorization\s*[:=]\s*)(?:bearer\s+)?[^\s,;]+" ).expect("redaction regex")),
+        "query" => QUERY.get_or_init(|| regex::Regex::new(r"(?i)([?&](?:token|access_token|api_token|password|backup_password|secret|credential|credentials|key_secret|application_secret|consumer_key)=)[^&\s]+" ).expect("redaction regex")),
+        "json" => JSON.get_or_init(|| regex::Regex::new(r#"(?i)(\"(?:authorization|proxy_authorization|api_token|access_token|application_secret|consumer_key|key_secret|backup_password|password|credential|credentials|secret|token|private_key|master_key)\"\s*:\s*\")[^\"]*(\")"#).expect("redaction regex")),
+        "kv" => KV.get_or_init(|| regex::Regex::new(r"(?i)((?:authorization|proxy[_-]?authorization|api[_-]?token|access[_-]?token|application[_-]?secret|consumer[_-]?key|key[_-]?secret|backup[_-]?password|password|credential|credentials|secret|token|private[_-]?key|master[_-]?key)\s*[=:]\s*)[^\s,;\"']+" ).expect("redaction regex")),
+        "pem" => PEM.get_or_init(|| regex::Regex::new(r"(?s)-----BEGIN [^-\r\n]*PRIVATE KEY-----.*?-----END [^-\r\n]*PRIVATE KEY-----" ).expect("redaction regex")),
+        _ => unreachable!(),
+    }
+}
+
+fn is_secret_field(name: &str) -> bool {
+    matches!(
+        name.to_ascii_lowercase().as_str(),
+        "authorization"
+            | "proxy_authorization"
+            | "api_token"
+            | "access_token"
+            | "application_secret"
+            | "consumer_key"
+            | "key_secret"
+            | "backup_password"
+            | "password"
+            | "credential"
+            | "credentials"
+            | "secret"
+            | "token"
+            | "private_key"
+            | "master_key"
+    )
+}
+
+pub fn redact_text(input: &str) -> String {
+    let mut value = redact_regex("pem").replace_all(input, "[REDACTED PRIVATE KEY]").into_owned();
+    value = redact_regex("auth").replace_all(&value, "$1[REDACTED]").into_owned();
+    value = redact_regex("query").replace_all(&value, "$1[REDACTED]").into_owned();
+    value = redact_regex("json").replace_all(&value, "$1[REDACTED]$2").into_owned();
+    redact_regex("kv").replace_all(&value, "$1[REDACTED]").into_owned()
+}
+
+fn redact_value(value: &mut Value) {
+    match value {
+        Value::String(text) => *text = redact_text(text),
+        Value::Array(items) => for item in items { redact_value(item); },
+        Value::Object(items) => {
+            for (key, item) in items {
+                if is_secret_field(key) {
+                    *item = Value::String("[REDACTED]".into());
+                } else {
+                    redact_value(item);
+                }
+            }
+        }
+        _ => {}
+    }
+}
+
+fn rfc3339_now() -> String {
+    let ms = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_millis() as i64;
+    let seconds = ms / 1000;
+    let millis = ms % 1000;
+    let days = seconds.div_euclid(86_400);
+    let day_seconds = seconds.rem_euclid(86_400);
+    let z = days + 719_468;
+    let era = if z >= 0 { z } else { z - 146_096 }.div_euclid(146_097);
+    let doe = z - era * 146_097;
+    let yoe = (doe - doe / 1_460 + doe / 36_524 - doe / 146_096) / 365;
+    let mut year = yoe + era * 400;
+    let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
+    let mp = (5 * doy + 2) / 153;
+    let day = doy - (153 * mp + 2) / 5 + 1;
+    let month = mp + if mp < 10 { 3 } else { -9 };
+    if month <= 2 { year += 1; }
+    let hour = day_seconds / 3600;
+    let minute = (day_seconds % 3600) / 60;
+    let second = day_seconds % 60;
+    format!("{year:04}-{month:02}-{day:02}T{hour:02}:{minute:02}:{second:02}.{millis:03}Z")
+}
+
+async fn open_error_file(path: PathBuf) -> Result<File> {
+    let file = tokio::task::spawn_blocking(move || -> Result<std::fs::File> {
+        let mut options = std::fs::OpenOptions::new();
+        options.create(true).append(true).write(true);
+        #[cfg(unix)]
+        {
+            use std::os::unix::fs::OpenOptionsExt;
+            options.mode(0o600).custom_flags(libc::O_NOFOLLOW);
+        }
+        let file = options.open(&path)?;
+        let metadata = file.metadata()?;
+        ensure!(metadata.is_file(), "error log path is not a regular file");
+        #[cfg(unix)]
+        {
+            use std::os::unix::fs::PermissionsExt;
+            file.set_permissions(std::fs::Permissions::from_mode(0o600))?;
+        }
+        Ok(file)
+    }).await.map_err(|_| anyhow::anyhow!("error log open worker failed"))??;
+    Ok(File::from_std(file))
+}
+
+fn format_error_event(event: &ErrorEvent, format: ErrorLogFormat) -> Vec<u8> {
+    match format {
+        ErrorLogFormat::Json => {
+            let mut object = Map::new();
+            object.insert("timestamp".into(), Value::String(event.timestamp.clone()));
+            object.insert("level".into(), Value::String("ERROR".into()));
+            object.insert("target".into(), Value::String(event.target.clone()));
+            object.insert("message".into(), Value::String(event.message.clone()));
+            for (key, value) in &event.fields {
+                if !matches!(key.as_str(), "timestamp" | "level" | "target" | "message") {
+                    object.insert(key.clone(), Value::String(value.clone()));
+                }
+            }
+            let mut line = serde_json::to_vec(&Value::Object(object)).unwrap_or_default();
+            line.push(b'\n');
+            line
+        }
+        ErrorLogFormat::Raw => {
+            let one_line = |value: &str| value.replace('\r', "\\r").replace('\n', "\\n");
+            let mut line = format!(
+                "{} ERROR {} {}",
+                event.timestamp,
+                event.target,
+                one_line(&event.message)
+            );
+            for (key, value) in &event.fields {
+                line.push(' ');
+                line.push_str(key);
+                line.push('=');
+                line.push_str(&one_line(value));
+            }
+            line.push('\n');
+            line.into_bytes()
+        }
+    }
+}
+
+async fn error_log_loop(
+    mut receiver: mpsc::Receiver<ErrorEvent>,
+    settings: Arc<ArcSwap<ErrorLogSettings>>,
+    mut settings_changed: watch::Receiver<()>,
+    written: Arc<AtomicU64>,
+    errors: Arc<AtomicU64>,
+) {
+    let mut open_path: Option<PathBuf> = None;
+    let mut file: Option<File> = None;
+    let mut retry_after: Option<Instant> = None;
+    let mut last_warning: Option<Instant> = None;
+    loop {
+        let event = tokio::select! {
+            event = receiver.recv() => match event {
+                Some(event) => event,
+                None => break,
+            },
+            changed = settings_changed.changed() => {
+                if changed.is_err() && receiver.is_closed() {
+                    break;
+                }
+                let config = settings.load_full();
+                let path = PathBuf::from(&config.path);
+                if !config.enabled || open_path.as_ref() != Some(&path) {
+                    // Dropping Tokio's File closes the descriptor without blocking
+                    // the request path. A changed path is reopened on the next event.
+                    file = None;
+                    open_path = None;
+                    retry_after = None;
+                }
+                continue;
+            }
+        };
+
+        let config = settings.load_full();
+        if !config.enabled {
+            file = None;
+            open_path = None;
+            retry_after = None;
+            continue;
+        }
+        let path = PathBuf::from(&config.path);
+        if open_path.as_ref() != Some(&path) {
+            file = None;
+            open_path = Some(path.clone());
+            retry_after = None;
+        }
+        if file.is_none() && retry_after.is_none_or(|until| Instant::now() >= until) {
+            match open_error_file(path.clone()).await {
+                Ok(opened) => {
+                    file = Some(opened);
+                    retry_after = None;
+                }
+                Err(error) => {
+                    errors.fetch_add(1, Ordering::Relaxed);
+                    retry_after = Some(Instant::now() + ERROR_LOG_RETRY);
+                    if last_warning.is_none_or(|time| time.elapsed() >= ERROR_LOG_RETRY) {
+                        tracing::warn!(path=%path.display(), %error, "optional error log file is unavailable");
+                        last_warning = Some(Instant::now());
+                    }
+                }
+            }
+        }
+        let Some(opened) = file.as_mut() else { continue; };
+        let line = format_error_event(&event, config.format);
+        if let Err(error) = opened.write_all(&line).await {
+            errors.fetch_add(1, Ordering::Relaxed);
+            file = None;
+            retry_after = Some(Instant::now() + ERROR_LOG_RETRY);
+            if last_warning.is_none_or(|time| time.elapsed() >= ERROR_LOG_RETRY) {
+                tracing::warn!(path=%path.display(), %error, "optional error log write failed");
+                last_warning = Some(Instant::now());
+            }
+        } else {
+            written.fetch_add(1, Ordering::Relaxed);
+        }
+    }
+}
+
 #[derive(Debug, Clone, Serialize)]
 pub struct LogEntry {
     pub vhost: Option<String>,
@@ -476,7 +884,8 @@ async fn sink_loop(
             entry = rx.recv() => match entry { Some(entry) => entry, None => break },
         };
         let config = settings.load_full();
-        let payload = format_entry(&entry, &config.format);
+        let mut payload = format_entry(&entry, &config.format);
+        redact_value(&mut payload);
         let mut attempted = false;
         let mut failed = false;
         if config.file.enabled {
@@ -716,6 +1125,38 @@ mod tests {
         assert_eq!(value["response_bytes"], 42);
     }
 
+    #[test]
+    fn error_log_defaults_off_and_redacts_common_secret_shapes() {
+        let settings = ErrorLogSettings::default();
+        assert!(!settings.enabled);
+        assert_eq!(settings.format, ErrorLogFormat::Json);
+        let text = redact_text(
+            r#"Authorization: Bearer topsecret url=https://x.invalid/?api_token=abc password=hunter2 {\"private_key\":\"pem-data\"}"#,
+        );
+        assert!(!text.contains("topsecret"));
+        assert!(!text.contains("api_token=abc"));
+        assert!(!text.contains("hunter2"));
+        assert!(!text.contains("pem-data"));
+    }
+
+    #[test]
+    fn error_json_is_one_record_and_reserved_fields_cannot_be_overwritten() {
+        let event = ErrorEvent {
+            timestamp: "2026-10-03T12:34:56.123Z".into(),
+            target: "rust_proxy::http".into(),
+            message: "upstream failed\nsecond line".into(),
+            fields: BTreeMap::from([
+                ("message".into(), "attacker override".into()),
+                ("request_id".into(), "abc".into()),
+            ]),
+        };
+        let line = format_error_event(&event, ErrorLogFormat::Json);
+        assert_eq!(line.iter().filter(|byte| **byte == b'\n').count(), 1);
+        let value: Value = serde_json::from_slice(&line).unwrap();
+        assert_eq!(value["message"], "upstream failed\nsecond line");
+        assert_eq!(value["request_id"], "abc");
+    }
+
     #[test]
     fn retained_history_is_larger_than_the_live_snapshot_but_still_bounded() {
         let logs = Logs::default();