1use 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
44pub struct JsonLinesProcessor;
46
47impl JsonLinesProcessor {
48 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 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 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 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 }
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 let store1 = make_store();
254 let proc = JsonLinesProcessor;
255 let from_process = proc.process(input, &profile, &store1).unwrap();
256
257 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 #[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}