rust_sanitize/processor/
jsonl_proc.rs

1//! NDJSON / JSON Lines structured processor.
2//!
3//! Processes files where each non-empty line is an independent JSON object
4//! (Newline-Delimited JSON, also called JSON Lines). Unlike the [`JsonProcessor`](crate::processor::json_proc::JsonProcessor),
5//! this processor never builds a full in-memory parse tree for the whole file —
6//! each line is parsed, walked, serialised, and written out independently,
7//! keeping per-line memory overhead constant regardless of input size.
8//!
9//! When used via the CLI with a matching profile, the processor is invoked
10//! through the streaming path: the file is opened as a reader and processed
11//! line-by-line without `fs::read` loading it into a `Vec<u8>` first. This
12//! makes GB-scale NDJSON log files practical to sanitize.
13//!
14//! # Options
15//!
16//! | Key | Values | Default | Description |
17//! |-----|--------|---------|-------------|
18//! | `skip_invalid` | `"true"` / `"false"` | `"false"` | Pass malformed lines through unchanged instead of returning an error. Useful for mixed log files that interleave plain-text lines with JSON. |
19//! | `compact` | `"true"` / `"false"` | `"true"` | Serialise each output line as compact JSON. Set to `"false"` only for debugging — pretty-printed NDJSON is non-standard. |
20//!
21//! # Example profile entry
22//!
23//! ```yaml
24//! - processor: jsonl
25//!   extensions: [".jsonl", ".ndjson", ".log"]
26//!   options:
27//!     skip_invalid: "true"
28//!   fields:
29//!     - pattern: "*.email"
30//!       category: email
31//!     - pattern: "*.password"
32//!       category: "custom:password"
33//! ```
34
35use crate::error::{Result, SanitizeError};
36use crate::processor::json_proc::walk_json;
37use crate::processor::limits::DEFAULT_INPUT_SIZE;
38use crate::processor::{FileTypeProfile, Processor};
39use crate::store::MappingStore;
40use serde_json::Value;
41use std::io::{self, BufRead, BufReader, Write};
42
43/// Structured processor for NDJSON / JSON Lines files.
44pub struct JsonLinesProcessor;
45
46impl JsonLinesProcessor {
47    /// Core line-by-line processing logic, shared by both `process` and
48    /// `process_stream`. Reads from any `BufRead` source and writes to any
49    /// `Write` sink.
50    fn process_lines(
51        reader: impl BufRead,
52        writer: &mut dyn Write,
53        profile: &FileTypeProfile,
54        store: &MappingStore,
55    ) -> Result<()> {
56        let skip_invalid = profile
57            .options
58            .get("skip_invalid")
59            .is_some_and(|v| v == "true");
60
61        let compact = profile.options.get("compact").is_none_or(|v| v != "false");
62
63        for (line_no, line_result) in reader.lines().enumerate() {
64            let raw_line = line_result?;
65
66            if raw_line.trim().is_empty() {
67                continue;
68            }
69
70            let mut value: Value = match serde_json::from_str(&raw_line) {
71                Ok(v) => v,
72                Err(e) => {
73                    if skip_invalid {
74                        writer.write_all(raw_line.as_bytes())?;
75                        writer.write_all(b"\n")?;
76                        continue;
77                    }
78                    return Err(SanitizeError::ParseError {
79                        format: "JSONL".into(),
80                        message: format!("line {}: {}", line_no + 1, e),
81                    });
82                }
83            };
84
85            walk_json(&mut value, "", profile, store, 0)?;
86
87            let serialised = if compact {
88                serde_json::to_vec(&value)
89            } else {
90                serde_json::to_vec_pretty(&value)
91            }
92            .map_err(|e| {
93                SanitizeError::IoError(std::io::Error::other(format!("JSONL serialize error: {e}")))
94            })?;
95
96            writer.write_all(&serialised)?;
97            writer.write_all(b"\n")?;
98        }
99
100        Ok(())
101    }
102}
103
104impl Processor for JsonLinesProcessor {
105    fn name(&self) -> &'static str {
106        "jsonl"
107    }
108
109    fn can_handle(&self, content: &[u8], profile: &FileTypeProfile) -> bool {
110        if profile.processor == "jsonl" {
111            return true;
112        }
113        // Heuristic: first non-empty line starts with `{` and there are
114        // multiple lines — distinguishes NDJSON from a single-object JSON file.
115        let Ok(text) = std::str::from_utf8(content) else {
116            return false;
117        };
118        let mut lines = text.lines().filter(|l| !l.trim().is_empty());
119        let first = match lines.next() {
120            Some(l) => l.trim_start(),
121            None => return false,
122        };
123        first.starts_with('{') && lines.next().is_some()
124    }
125
126    fn supports_streaming(&self) -> bool {
127        true
128    }
129
130    fn process(
131        &self,
132        content: &[u8],
133        profile: &FileTypeProfile,
134        store: &MappingStore,
135    ) -> Result<Vec<u8>> {
136        // Validate size and UTF-8 upfront so the error points at the file.
137        crate::processor::check_size_and_decode(content, "JSONL", DEFAULT_INPUT_SIZE)?;
138        let mut output = Vec::with_capacity(content.len());
139        Self::process_lines(BufReader::new(content), &mut output, profile, store)?;
140        Ok(output)
141    }
142
143    fn process_stream(
144        &self,
145        reader: &mut dyn io::Read,
146        writer: &mut dyn io::Write,
147        profile: &FileTypeProfile,
148        store: &MappingStore,
149    ) -> Result<()> {
150        Self::process_lines(BufReader::new(reader), writer, profile, store)
151    }
152}
153
154#[cfg(test)]
155mod tests {
156    use super::*;
157    use crate::category::Category;
158    use crate::generator::HmacGenerator;
159    use crate::processor::profile::FieldRule;
160    use std::sync::Arc;
161
162    fn make_store() -> MappingStore {
163        let gen = Arc::new(HmacGenerator::new([42u8; 32]));
164        MappingStore::new(gen, None)
165    }
166
167    fn make_profile(fields: Vec<FieldRule>) -> FileTypeProfile {
168        FileTypeProfile::new("jsonl", fields).with_option("compact", "true")
169    }
170
171    #[test]
172    fn replaces_matched_fields_across_lines() {
173        let store = make_store();
174        let proc = JsonLinesProcessor;
175        let input = b"{\"email\":\"a@b.com\",\"level\":\"info\"}\n{\"email\":\"c@d.com\",\"level\":\"warn\"}\n";
176        let profile = make_profile(vec![FieldRule::new("email").with_category(Category::Email)]);
177
178        let result = proc.process(input, &profile, &store).unwrap();
179        let lines: Vec<&str> = std::str::from_utf8(&result).unwrap().lines().collect();
180
181        assert_eq!(lines.len(), 2);
182        let v0: Value = serde_json::from_str(lines[0]).unwrap();
183        let v1: Value = serde_json::from_str(lines[1]).unwrap();
184        assert_ne!(v0["email"].as_str().unwrap(), "a@b.com");
185        assert_ne!(v1["email"].as_str().unwrap(), "c@d.com");
186        assert_eq!(v0["level"].as_str().unwrap(), "info");
187        assert_eq!(v1["level"].as_str().unwrap(), "warn");
188    }
189
190    #[test]
191    fn process_stream_matches_process() {
192        let input = b"{\"email\":\"a@b.com\"}\n{\"email\":\"c@d.com\"}\n";
193        let profile = make_profile(vec![FieldRule::new("email").with_category(Category::Email)]);
194
195        // process
196        let store1 = make_store();
197        let proc = JsonLinesProcessor;
198        let from_process = proc.process(input, &profile, &store1).unwrap();
199
200        // process_stream with identical store seed
201        let store2 = make_store();
202        let mut reader = io::Cursor::new(input);
203        let mut from_stream = Vec::new();
204        proc.process_stream(&mut reader, &mut from_stream, &profile, &store2)
205            .unwrap();
206
207        assert_eq!(from_process, from_stream);
208    }
209
210    #[test]
211    fn glob_suffix_pattern() {
212        let store = make_store();
213        let proc = JsonLinesProcessor;
214        let input = b"{\"db\":{\"password\":\"pw1\"},\"name\":\"app\"}\n";
215        let profile = make_profile(vec![
216            FieldRule::new("*.password").with_category(Category::Custom("pw".into()))
217        ]);
218
219        let result = proc.process(input, &profile, &store).unwrap();
220        let trimmed = result
221            .iter()
222            .rposition(|b| !b.is_ascii_whitespace())
223            .map_or(&[][..], |i| &result[..=i]);
224        let v: Value = serde_json::from_slice(trimmed).unwrap();
225        assert_ne!(v["db"]["password"].as_str().unwrap(), "pw1");
226        assert_eq!(v["name"].as_str().unwrap(), "app");
227    }
228
229    #[test]
230    fn skip_invalid_passes_through_bad_lines() {
231        let store = make_store();
232        let proc = JsonLinesProcessor;
233        let input = b"{\"email\":\"a@b.com\"}\nnot json at all\n{\"email\":\"c@d.com\"}\n";
234        let profile = FileTypeProfile::new(
235            "jsonl",
236            vec![FieldRule::new("email").with_category(Category::Email)],
237        )
238        .with_option("skip_invalid", "true")
239        .with_option("compact", "true");
240
241        let result = proc.process(input, &profile, &store).unwrap();
242        let text = std::str::from_utf8(&result).unwrap();
243        let lines: Vec<&str> = text.lines().collect();
244
245        assert_eq!(lines.len(), 3);
246        assert_eq!(lines[1], "not json at all");
247        let v0: Value = serde_json::from_str(lines[0]).unwrap();
248        assert_ne!(v0["email"].as_str().unwrap(), "a@b.com");
249    }
250
251    #[test]
252    fn error_on_invalid_line_by_default() {
253        let store = make_store();
254        let proc = JsonLinesProcessor;
255        let input = b"{\"email\":\"a@b.com\"}\nnot json\n";
256        let profile = make_profile(vec![FieldRule::new("email").with_category(Category::Email)]);
257
258        assert!(proc.process(input, &profile, &store).is_err());
259    }
260
261    #[test]
262    fn deterministic_same_value_same_replacement() {
263        let store = make_store();
264        let proc = JsonLinesProcessor;
265        let input = b"{\"email\":\"a@b.com\"}\n{\"email\":\"a@b.com\"}\n";
266        let profile = make_profile(vec![FieldRule::new("email").with_category(Category::Email)]);
267
268        let result = proc.process(input, &profile, &store).unwrap();
269        let lines: Vec<&str> = std::str::from_utf8(&result).unwrap().lines().collect();
270        let v0: Value = serde_json::from_str(lines[0]).unwrap();
271        let v1: Value = serde_json::from_str(lines[1]).unwrap();
272        assert_eq!(v0["email"].as_str().unwrap(), v1["email"].as_str().unwrap());
273    }
274
275    #[test]
276    fn can_handle_heuristic_multi_line_json_objects() {
277        let proc = JsonLinesProcessor;
278        let profile = FileTypeProfile::new("yaml", vec![]);
279        let input = b"{\"a\":1}\n{\"b\":2}\n";
280        assert!(proc.can_handle(input, &profile));
281    }
282
283    #[test]
284    fn can_handle_rejects_single_object() {
285        let proc = JsonLinesProcessor;
286        let profile = FileTypeProfile::new("yaml", vec![]);
287        let input = b"{\"a\":1}";
288        assert!(!proc.can_handle(input, &profile));
289    }
290
291    #[test]
292    fn supports_streaming_is_true() {
293        assert!(JsonLinesProcessor.supports_streaming());
294    }
295}
rust_sanitize/processor/jsonl_proc.rs

rust_sanitize/processor/
jsonl_proc.rs
