scour_secrets/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).
5//!
6//! The CLI uses [`process_to_edits`](JsonLinesProcessor::process_to_edits): it
7//! runs the JSON span walker on each line and offsets the resulting edits by the
8//! line's byte position, so each line's exact formatting is preserved and values
9//! escaped in the source never leak. Files within `--max-structured-size` are
10//! processed this way; larger files fall back to the bounded-memory streaming
11//! path ([`process_stream`](JsonLinesProcessor::process_stream)), which parses,
12//! walks, and re-serializes each line independently — keeping per-line memory
13//! constant for GB-scale NDJSON.
14//!
15//! # Options
16//!
17//! | Key | Values | Default | Description |
18//! |-----|--------|---------|-------------|
19//! | `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. |
20//! | `compact` | `"true"` / `"false"` | `"true"` | Serialise each output line as compact JSON. Set to `"false"` only for debugging — pretty-printed NDJSON is non-standard. |
21//!
22//! # Example profile entry
23//!
24//! ```yaml
25//! - processor: jsonl
26//!   extensions: [".jsonl", ".ndjson", ".log"]
27//!   options:
28//!     skip_invalid: "true"
29//!   fields:
30//!     - pattern: "*.email"
31//!       category: email
32//!     - pattern: "*.password"
33//!       category: "custom:password"
34//! ```
35
36use crate::error::{Result, SanitizeError};
37use crate::processor::json_proc::walk_json;
38use crate::processor::limits::DEFAULT_INPUT_SIZE;
39use crate::processor::{FileTypeProfile, Processor};
40use crate::store::MappingStore;
41use serde_json::Value;
42use std::io::{self, BufRead, BufReader, Write};
43
44/// Structured processor for NDJSON / JSON Lines files.
45pub struct JsonLinesProcessor;
46
47impl JsonLinesProcessor {
48    /// Core line-by-line processing logic, shared by both `process` and
49    /// `process_stream`. Reads from any `BufRead` source and writes to any
50    /// `Write` sink.
51    fn process_lines(
52        reader: impl BufRead,
53        writer: &mut dyn Write,
54        profile: &FileTypeProfile,
55        store: &MappingStore,
56    ) -> Result<()> {
57        let skip_invalid = profile
58            .options
59            .get("skip_invalid")
60            .is_some_and(|v| v == "true");
61
62        let compact = profile.options.get("compact").is_none_or(|v| v != "false");
63
64        for (line_no, line_result) in reader.lines().enumerate() {
65            let raw_line = line_result?;
66
67            if raw_line.trim().is_empty() {
68                continue;
69            }
70
71            let mut value: Value = match serde_json::from_str(&raw_line) {
72                Ok(v) => v,
73                Err(e) => {
74                    if skip_invalid {
75                        writer.write_all(raw_line.as_bytes())?;
76                        writer.write_all(b"\n")?;
77                        continue;
78                    }
79                    return Err(SanitizeError::ParseError {
80                        format: "JSONL".into(),
81                        message: format!("line {}: {}", line_no + 1, e),
82                    });
83                }
84            };
85
86            walk_json(&mut value, "", profile, store, 0)?;
87
88            let serialised = if compact {
89                serde_json::to_vec(&value)
90            } else {
91                serde_json::to_vec_pretty(&value)
92            }
93            .map_err(|e| {
94                SanitizeError::IoError(std::io::Error::other(format!("JSONL serialize error: {e}")))
95            })?;
96
97            writer.write_all(&serialised)?;
98            writer.write_all(b"\n")?;
99        }
100
101        Ok(())
102    }
103}
104
105impl Processor for JsonLinesProcessor {
106    fn name(&self) -> &'static str {
107        "jsonl"
108    }
109
110    fn can_handle(&self, content: &[u8], profile: &FileTypeProfile) -> bool {
111        if profile.processor == "jsonl" {
112            return true;
113        }
114        // Heuristic: first non-empty line starts with `{` and there are
115        // multiple lines — distinguishes NDJSON from a single-object JSON file.
116        let Ok(text) = std::str::from_utf8(content) else {
117            return false;
118        };
119        let mut lines = text.lines().filter(|l| !l.trim().is_empty());
120        let first = match lines.next() {
121            Some(l) => l.trim_start(),
122            None => return false,
123        };
124        first.starts_with('{') && lines.next().is_some()
125    }
126
127    fn supports_streaming(&self) -> bool {
128        true
129    }
130
131    fn process(
132        &self,
133        content: &[u8],
134        profile: &FileTypeProfile,
135        store: &MappingStore,
136    ) -> Result<Vec<u8>> {
137        // Validate size and UTF-8 upfront so the error points at the file.
138        crate::processor::check_size_and_decode(content, "JSONL", DEFAULT_INPUT_SIZE)?;
139        let mut output = Vec::with_capacity(content.len());
140        Self::process_lines(BufReader::new(content), &mut output, profile, store)?;
141        Ok(output)
142    }
143
144    /// Span-based redaction: run the JSON span walker on each line, offsetting
145    /// the resulting edits by the line's byte position. Preserves each line's
146    /// exact formatting; invalid lines are passed through (no edits) when
147    /// `skip_invalid` is set, matching `process`.
148    fn process_to_edits(
149        &self,
150        content: &[u8],
151        profile: &FileTypeProfile,
152        store: &MappingStore,
153    ) -> Result<Option<Vec<crate::processor::Replacement>>> {
154        crate::processor::check_size_and_decode(content, "JSONL", DEFAULT_INPUT_SIZE)?;
155        let skip_invalid = profile
156            .options
157            .get("skip_invalid")
158            .is_some_and(|v| v == "true");
159
160        let mut edits = Vec::new();
161        let mut pos = 0usize;
162        let mut line_no = 0usize;
163        while pos < content.len() {
164            let rel_nl = content[pos..].iter().position(|&b| b == b'\n');
165            let end = rel_nl.map_or(content.len(), |i| pos + i);
166            let line = &content[pos..end];
167            line_no += 1;
168
169            if !line.iter().all(u8::is_ascii_whitespace) {
170                match crate::processor::json_proc::json_value_edits(line, profile, store) {
171                    Ok(line_edits) => {
172                        for e in line_edits {
173                            edits.push(crate::processor::Replacement {
174                                start: e.start + pos,
175                                end: e.end + pos,
176                                value: e.value,
177                            });
178                        }
179                    }
180                    Err(e) => {
181                        if !skip_invalid {
182                            return Err(SanitizeError::ParseError {
183                                format: "JSONL".into(),
184                                message: format!("line {line_no}: {e}"),
185                            });
186                        }
187                        // skip_invalid: leave the line unchanged (no edits).
188                    }
189                }
190            }
191
192            match rel_nl {
193                Some(_) => pos = end + 1,
194                None => break,
195            }
196        }
197        Ok(Some(edits))
198    }
199
200    fn process_stream(
201        &self,
202        reader: &mut dyn io::Read,
203        writer: &mut dyn io::Write,
204        profile: &FileTypeProfile,
205        store: &MappingStore,
206    ) -> Result<()> {
207        Self::process_lines(BufReader::new(reader), writer, profile, store)
208    }
209}
210
211#[cfg(test)]
212mod tests {
213    use super::*;
214    use crate::category::Category;
215    use crate::generator::HmacGenerator;
216    use crate::processor::profile::FieldRule;
217    use std::sync::Arc;
218
219    fn make_store() -> MappingStore {
220        let gen = Arc::new(HmacGenerator::new([42u8; 32]));
221        MappingStore::new(gen, None)
222    }
223
224    fn make_profile(fields: Vec<FieldRule>) -> FileTypeProfile {
225        FileTypeProfile::new("jsonl", fields).with_option("compact", "true")
226    }
227
228    #[test]
229    fn replaces_matched_fields_across_lines() {
230        let store = make_store();
231        let proc = JsonLinesProcessor;
232        let input = b"{\"email\":\"a@b.com\",\"level\":\"info\"}\n{\"email\":\"c@d.com\",\"level\":\"warn\"}\n";
233        let profile = make_profile(vec![FieldRule::new("email").with_category(Category::Email)]);
234
235        let result = proc.process(input, &profile, &store).unwrap();
236        let lines: Vec<&str> = std::str::from_utf8(&result).unwrap().lines().collect();
237
238        assert_eq!(lines.len(), 2);
239        let v0: Value = serde_json::from_str(lines[0]).unwrap();
240        let v1: Value = serde_json::from_str(lines[1]).unwrap();
241        assert_ne!(v0["email"].as_str().unwrap(), "a@b.com");
242        assert_ne!(v1["email"].as_str().unwrap(), "c@d.com");
243        assert_eq!(v0["level"].as_str().unwrap(), "info");
244        assert_eq!(v1["level"].as_str().unwrap(), "warn");
245    }
246
247    #[test]
248    fn process_stream_matches_process() {
249        let input = b"{\"email\":\"a@b.com\"}\n{\"email\":\"c@d.com\"}\n";
250        let profile = make_profile(vec![FieldRule::new("email").with_category(Category::Email)]);
251
252        // process
253        let store1 = make_store();
254        let proc = JsonLinesProcessor;
255        let from_process = proc.process(input, &profile, &store1).unwrap();
256
257        // process_stream with identical store seed
258        let store2 = make_store();
259        let mut reader = io::Cursor::new(input);
260        let mut from_stream = Vec::new();
261        proc.process_stream(&mut reader, &mut from_stream, &profile, &store2)
262            .unwrap();
263
264        assert_eq!(from_process, from_stream);
265    }
266
267    #[test]
268    fn glob_suffix_pattern() {
269        let store = make_store();
270        let proc = JsonLinesProcessor;
271        let input = b"{\"db\":{\"password\":\"pw1\"},\"name\":\"app\"}\n";
272        let profile = make_profile(vec![
273            FieldRule::new("*.password").with_category(Category::Custom("pw".into()))
274        ]);
275
276        let result = proc.process(input, &profile, &store).unwrap();
277        let trimmed = result
278            .iter()
279            .rposition(|b| !b.is_ascii_whitespace())
280            .map_or(&[][..], |i| &result[..=i]);
281        let v: Value = serde_json::from_slice(trimmed).unwrap();
282        assert_ne!(v["db"]["password"].as_str().unwrap(), "pw1");
283        assert_eq!(v["name"].as_str().unwrap(), "app");
284    }
285
286    #[test]
287    fn skip_invalid_passes_through_bad_lines() {
288        let store = make_store();
289        let proc = JsonLinesProcessor;
290        let input = b"{\"email\":\"a@b.com\"}\nnot json at all\n{\"email\":\"c@d.com\"}\n";
291        let profile = FileTypeProfile::new(
292            "jsonl",
293            vec![FieldRule::new("email").with_category(Category::Email)],
294        )
295        .with_option("skip_invalid", "true")
296        .with_option("compact", "true");
297
298        let result = proc.process(input, &profile, &store).unwrap();
299        let text = std::str::from_utf8(&result).unwrap();
300        let lines: Vec<&str> = text.lines().collect();
301
302        assert_eq!(lines.len(), 3);
303        assert_eq!(lines[1], "not json at all");
304        let v0: Value = serde_json::from_str(lines[0]).unwrap();
305        assert_ne!(v0["email"].as_str().unwrap(), "a@b.com");
306    }
307
308    #[test]
309    fn error_on_invalid_line_by_default() {
310        let store = make_store();
311        let proc = JsonLinesProcessor;
312        let input = b"{\"email\":\"a@b.com\"}\nnot json\n";
313        let profile = make_profile(vec![FieldRule::new("email").with_category(Category::Email)]);
314
315        assert!(proc.process(input, &profile, &store).is_err());
316    }
317
318    #[test]
319    fn deterministic_same_value_same_replacement() {
320        let store = make_store();
321        let proc = JsonLinesProcessor;
322        let input = b"{\"email\":\"a@b.com\"}\n{\"email\":\"a@b.com\"}\n";
323        let profile = make_profile(vec![FieldRule::new("email").with_category(Category::Email)]);
324
325        let result = proc.process(input, &profile, &store).unwrap();
326        let lines: Vec<&str> = std::str::from_utf8(&result).unwrap().lines().collect();
327        let v0: Value = serde_json::from_str(lines[0]).unwrap();
328        let v1: Value = serde_json::from_str(lines[1]).unwrap();
329        assert_eq!(v0["email"].as_str().unwrap(), v1["email"].as_str().unwrap());
330    }
331
332    #[test]
333    fn can_handle_heuristic_multi_line_json_objects() {
334        let proc = JsonLinesProcessor;
335        let profile = FileTypeProfile::new("yaml", vec![]);
336        let input = b"{\"a\":1}\n{\"b\":2}\n";
337        assert!(proc.can_handle(input, &profile));
338    }
339
340    #[test]
341    fn can_handle_rejects_single_object() {
342        let proc = JsonLinesProcessor;
343        let profile = FileTypeProfile::new("yaml", vec![]);
344        let input = b"{\"a\":1}";
345        assert!(!proc.can_handle(input, &profile));
346    }
347
348    #[test]
349    fn supports_streaming_is_true() {
350        assert!(JsonLinesProcessor.supports_streaming());
351    }
352
353    /// Edit-mode redacts matched values per line (including non-canonical `\/`
354    /// escapes), offsets spans correctly, and leaves invalid lines unchanged
355    /// when `skip_invalid` is set.
356    #[test]
357    fn edits_redact_per_line_and_skip_invalid() {
358        let store = make_store();
359        let proc = JsonLinesProcessor;
360        let content =
361            b"{\"email\":\"a-SEC1@e.test\"}\n{\"u\":\"http:\\/\\/SEC2.test\"}\nnot json\n";
362        let mut profile = FileTypeProfile::new(
363            "jsonl",
364            vec![
365                FieldRule::new("email").with_category(Category::Email),
366                FieldRule::new("u").with_category(Category::Custom("url".into())),
367            ],
368        );
369        profile.options.insert("skip_invalid".into(), "true".into());
370        let edits = proc
371            .process_to_edits(content, &profile, &store)
372            .unwrap()
373            .unwrap();
374        let out = crate::processor::apply_edits(content, edits);
375        let text = String::from_utf8(out).unwrap();
376        assert!(!text.contains("SEC1"), "leaked SEC1: {text}");
377        assert!(!text.contains("SEC2"), "leaked SEC2 (\\/ escape): {text}");
378        assert!(text.contains("not json"), "invalid line dropped: {text}");
379    }
380}
scour_secrets/processor/jsonl_proc.rs

scour_secrets/processor/
jsonl_proc.rs
