document_svg/document/
cloudevents.rs

1//! Bounded CloudEvents JSON 1.0 previews.
2//!
3//! CloudEvents JSON envelopes carry event metadata and an optional data payload.
4//! This adapter renders only inert metadata (with URI query values masked): data,
5//! extension values and external schemas are summarized without being displayed,
6//! decoded, fetched, or executed.
7
8use serde_json::Value;
9use std::path::Path;
10
11use crate::convert::{ConvertOptions, PageConsumer, read_limited_file};
12use crate::document::html::{HtmlBlock, render_blocks_to_pages};
13use crate::error::{Error, Result};
14use crate::table::{TableAlign, TableData};
15
16const MAX_CLOUDEVENTS_BYTES: u64 = 64 * 1024 * 1024;
17const MAX_CLOUDEVENTS_DEPTH: usize = 100;
18const MAX_CLOUDEVENTS_VALUES: usize = 300_000;
19const MAX_CLOUDEVENTS_EVENTS: usize = 100_000;
20const MAX_CLOUDEVENTS_STRING_BYTES: usize = 2 * 1024 * 1024;
21const MAX_CLOUDEVENTS_DISPLAY_BYTES: usize = 512;
22const MAX_CLOUDEVENTS_URI_BYTES: usize = 512;
23
24/// Detect a CloudEvents JSON object or batch from a bounded prefix.
25pub(crate) fn looks_like_prefix(prefix: &[u8]) -> bool {
26    let text = String::from_utf8_lossy(prefix);
27    let trimmed = text.trim_start_matches('\u{feff}').trim_start();
28    if !(trimmed.starts_with('{') || trimmed.starts_with('[')) {
29        return false;
30    }
31    if let Ok(value) = serde_json::from_str::<Value>(trimmed) {
32        return match value {
33            Value::Object(object) => looks_like_event_object(&object),
34            Value::Array(items) => items
35                .first()
36                .and_then(Value::as_object)
37                .is_some_and(looks_like_event_object),
38            _ => false,
39        };
40    }
41    text.contains("\"specversion\"")
42        && text.contains("\"type\"")
43        && text.contains("\"source\"")
44        && text.contains("\"id\"")
45}
46
47fn looks_like_event_object(object: &serde_json::Map<String, Value>) -> bool {
48    ["specversion", "type", "source", "id"]
49        .iter()
50        .all(|key| object.get(*key).and_then(Value::as_str).is_some())
51}
52
53struct CloudEventsPageSink<'a> {
54    inner: &'a mut dyn PageConsumer,
55    warnings: &'a [String],
56}
57
58impl PageConsumer for CloudEventsPageSink<'_> {
59    fn consume(&mut self, mut page: crate::ir::Page) -> Result<()> {
60        page.source_format = "cloudevents".into();
61        if page.title.is_empty() {
62            page.title = "CloudEvents JSON".into();
63        }
64        page.description =
65            "CloudEvents envelope metadata is rendered inertly; data, extension values, and external schemas are not displayed or resolved".into();
66        for warning in self.warnings {
67            page.warn(warning.clone());
68        }
69        self.inner.consume(page)
70    }
71}
72
73pub(crate) fn convert(
74    path: &Path,
75    options: &ConvertOptions,
76    sink: &mut dyn PageConsumer,
77) -> Result<Vec<String>> {
78    let bytes = read_limited_file(
79        path,
80        options.max_input_bytes.min(MAX_CLOUDEVENTS_BYTES),
81        "CloudEvents JSON input",
82    )?;
83    let text = String::from_utf8(bytes).map_err(|error| {
84        Error::InvalidInput(format!("CloudEvents JSON must be UTF-8 JSON: {error}"))
85    })?;
86    let (table, metadata, warnings) = parse(&text)?;
87    let blocks = vec![
88        HtmlBlock::Heading {
89            level: 1,
90            text: "CloudEvents JSON".into(),
91        },
92        HtmlBlock::Paragraph { text: metadata },
93        HtmlBlock::Table(table),
94    ];
95    let mut page_sink = CloudEventsPageSink {
96        inner: sink,
97        warnings: &warnings,
98    };
99    render_blocks_to_pages(&blocks, &mut page_sink, options)?;
100    Ok(warnings)
101}
102
103fn parse(text: &str) -> Result<(TableData, String, Vec<String>)> {
104    if text.len() as u64 > MAX_CLOUDEVENTS_BYTES {
105        return Err(Error::LimitExceeded(format!(
106            "CloudEvents JSON exceeds {MAX_CLOUDEVENTS_BYTES} bytes"
107        )));
108    }
109    preflight_depth(text)?;
110    let value: Value = serde_json::from_str(text)
111        .map_err(|error| Error::InvalidInput(format!("invalid CloudEvents JSON: {error}")))?;
112    let mut values = 0usize;
113    count_values(&value, 0, &mut values)?;
114    let events: Vec<&Value> = match &value {
115        Value::Object(_) => vec![&value],
116        Value::Array(items) => {
117            if items.len() > MAX_CLOUDEVENTS_EVENTS {
118                return Err(Error::LimitExceeded(format!(
119                    "CloudEvents batch exceeds {MAX_CLOUDEVENTS_EVENTS} events"
120                )));
121            }
122            items.iter().collect()
123        }
124        _ => {
125            return Err(Error::InvalidInput(
126                "CloudEvents JSON root must be an event object or batch array".into(),
127            ));
128        }
129    };
130    let mut rows = Vec::with_capacity(events.len().min(MAX_CLOUDEVENTS_EVENTS));
131    let mut warnings = Vec::new();
132    let mut data_count = 0usize;
133    let mut base64_count = 0usize;
134    let mut schema_count = 0usize;
135    let mut extension_count = 0usize;
136    let mut content_types = std::collections::BTreeSet::new();
137    let mut specversions = std::collections::BTreeSet::new();
138    for (index, event) in events.iter().enumerate() {
139        let object = event.as_object().ok_or_else(|| {
140            Error::InvalidInput(format!(
141                "CloudEvents batch item {} must be an object",
142                index + 1
143            ))
144        })?;
145        let specversion = required_string(object, "specversion", index)?;
146        if specversion != "1.0" {
147            return Err(Error::Unsupported(format!(
148                "CloudEvents specversion '{specversion}' is unsupported (expected 1.0)"
149            )));
150        }
151        specversions.insert(specversion.to_owned());
152        let event_type = required_string(object, "type", index)?;
153        let source = required_string(object, "source", index)?;
154        let id = required_string(object, "id", index)?;
155        let time = optional_string(object, "time", index)?.map_or_else(|| "—".into(), truncate);
156        let subject =
157            optional_string(object, "subject", index)?.map_or_else(|| "—".into(), truncate);
158        let content_type =
159            optional_string(object, "datacontenttype", index)?.map_or_else(|| "—".into(), truncate);
160        if content_type != "—" && content_type != "(invalid)" {
161            content_types.insert(content_type.clone());
162        }
163        if object.contains_key("data") && object.contains_key("data_base64") {
164            return Err(Error::InvalidInput(format!(
165                "CloudEvents batch item {} cannot contain both data and data_base64",
166                index + 1
167            )));
168        }
169        let data = if let Some(data) = object.get("data") {
170            data_count = data_count.saturating_add(1);
171            format!(
172                "{}/{}B",
173                json_type(data),
174                serde_json::to_string(data).map_or(0, |s| s.len())
175            )
176        } else if let Some(encoded) = object.get("data_base64") {
177            let encoded = encoded.as_str().ok_or_else(|| {
178                Error::InvalidInput(format!(
179                    "CloudEvents batch item {} data_base64 must be a string",
180                    index + 1
181                ))
182            })?;
183            base64_count = base64_count.saturating_add(1);
184            format!("base64 ({} chars)", encoded.len())
185        } else {
186            "—".into()
187        };
188        if object.contains_key("dataschema") {
189            optional_string(object, "dataschema", index)?;
190            schema_count = schema_count.saturating_add(1);
191        }
192        let extension = object
193            .keys()
194            .filter(|key| !STANDARD_ATTRIBUTES.contains(&key.as_str()))
195            .count();
196        extension_count = extension_count.saturating_add(extension);
197        rows.push(vec![
198            truncate(event_type),
199            truncate(id),
200            format!("{} · {} · {} · {data}", source_label(source), time, subject),
201        ]);
202    }
203    if rows.is_empty() {
204        return Err(Error::InvalidInput(
205            "CloudEvents JSON batch must contain at least one event".into(),
206        ));
207    }
208    let metadata = format!(
209        "Events: {}\nSpecversion: {}\nWith data: {}\nWith data_base64: {}\nWith dataschema: {}\nExtension attributes: {}",
210        rows.len(),
211        specversions.into_iter().collect::<Vec<_>>().join(", "),
212        data_count,
213        base64_count,
214        schema_count,
215        extension_count
216    );
217    let metadata = format!(
218        "{metadata}\nContent types: {}",
219        if content_types.is_empty() {
220            "—".into()
221        } else {
222            content_types.into_iter().collect::<Vec<_>>().join(", ")
223        }
224    );
225    warnings.push("CloudEvents data/data_base64 payloads are summarized by type and size only; payloads are never decoded, rendered, executed or sent to a network endpoint".into());
226    warnings.push("CloudEvents source and dataschema URI values are displayed inertly with query values masked; no URI is fetched or resolved".into());
227    if extension_count > 0 {
228        warnings.push(format!(
229            "{extension_count} CloudEvents extension attribute(s) were counted but their values were omitted"
230        ));
231    }
232    Ok((
233        TableData {
234            headers: vec![
235                "Type".into(),
236                "ID".into(),
237                "Context (source · time · subject · data)".into(),
238            ],
239            rows,
240            alignments: vec![TableAlign::Left; 3],
241            raw_source: String::new(),
242        },
243        metadata,
244        warnings,
245    ))
246}
247
248const STANDARD_ATTRIBUTES: &[&str] = &[
249    "specversion",
250    "type",
251    "source",
252    "subject",
253    "id",
254    "time",
255    "datacontenttype",
256    "dataschema",
257    "data",
258    "data_base64",
259];
260
261fn required_string<'a>(
262    object: &'a serde_json::Map<String, Value>,
263    key: &str,
264    index: usize,
265) -> Result<&'a str> {
266    let value = object.get(key).and_then(Value::as_str).ok_or_else(|| {
267        Error::InvalidInput(format!(
268            "CloudEvents batch item {} requires string {key}",
269            index + 1
270        ))
271    })?;
272    if value.is_empty() {
273        return Err(Error::InvalidInput(format!(
274            "CloudEvents batch item {} {key} must not be empty",
275            index + 1
276        )));
277    }
278    Ok(value)
279}
280
281fn optional_string<'a>(
282    object: &'a serde_json::Map<String, Value>,
283    key: &str,
284    index: usize,
285) -> Result<Option<&'a str>> {
286    let Some(value) = object.get(key) else {
287        return Ok(None);
288    };
289    let value = value.as_str().ok_or_else(|| {
290        Error::InvalidInput(format!(
291            "CloudEvents batch item {} {key} must be a string",
292            index + 1
293        ))
294    })?;
295    if value.is_empty() {
296        return Err(Error::InvalidInput(format!(
297            "CloudEvents batch item {} {key} must not be empty",
298            index + 1
299        )));
300    }
301    Ok(Some(value))
302}
303
304fn json_type(value: &Value) -> &'static str {
305    match value {
306        Value::Null => "null",
307        Value::Bool(_) => "boolean",
308        Value::Number(_) => "number",
309        Value::String(_) => "string",
310        Value::Array(_) => "array",
311        Value::Object(_) => "object",
312    }
313}
314
315fn truncate(value: &str) -> String {
316    if value.len() <= MAX_CLOUDEVENTS_DISPLAY_BYTES {
317        return value.to_owned();
318    }
319    let mut end = MAX_CLOUDEVENTS_DISPLAY_BYTES;
320    while !value.is_char_boundary(end) {
321        end -= 1;
322    }
323    format!("{}…", &value[..end])
324}
325
326fn mask_uri(value: &str) -> String {
327    let mut masked = value.to_owned();
328    if let Some(query_start) = masked.find('?') {
329        let fragment_start = masked[query_start..]
330            .find('#')
331            .map_or(masked.len(), |offset| query_start + offset);
332        let query = &masked[query_start + 1..fragment_start];
333        let pairs = query
334            .split('&')
335            .filter(|pair| !pair.is_empty())
336            .map(|pair| {
337                let key = pair.split('=').next().unwrap_or("");
338                if key.is_empty() {
339                    "[redacted]".to_owned()
340                } else {
341                    format!("{key}=[redacted]")
342                }
343            })
344            .collect::<Vec<_>>();
345        masked.replace_range(query_start + 1..fragment_start, &pairs.join("&"));
346    }
347    truncate(
348        &masked
349            .chars()
350            .take(MAX_CLOUDEVENTS_URI_BYTES)
351            .collect::<String>(),
352    )
353}
354
355fn source_label(value: &str) -> String {
356    let masked = mask_uri(value);
357    let Some(scheme_end) = masked.find("://") else {
358        return truncate(&masked);
359    };
360    let authority_start = scheme_end + 3;
361    let authority_end = masked[authority_start..]
362        .find(['/', '?', '#'])
363        .map_or(masked.len(), |offset| authority_start + offset);
364    let authority = &masked[authority_start..authority_end];
365    if authority.is_empty() {
366        return truncate(&masked);
367    }
368    let mut label = authority.to_owned();
369    if masked[authority_end..].contains('?') {
370        label.push_str("?…");
371    }
372    truncate(&label)
373}
374
375fn count_values(value: &Value, depth: usize, count: &mut usize) -> Result<()> {
376    if depth > MAX_CLOUDEVENTS_DEPTH {
377        return Err(Error::LimitExceeded(format!(
378            "CloudEvents JSON nesting exceeds {MAX_CLOUDEVENTS_DEPTH} levels"
379        )));
380    }
381    *count = count.saturating_add(1);
382    if *count > MAX_CLOUDEVENTS_VALUES {
383        return Err(Error::LimitExceeded(format!(
384            "CloudEvents JSON contains more than {MAX_CLOUDEVENTS_VALUES} values"
385        )));
386    }
387    match value {
388        Value::Array(values) => {
389            for item in values {
390                count_values(item, depth + 1, count)?;
391            }
392        }
393        Value::Object(map) => {
394            for item in map.values() {
395                count_values(item, depth + 1, count)?;
396            }
397        }
398        Value::String(value) if value.len() > MAX_CLOUDEVENTS_STRING_BYTES => {
399            return Err(Error::LimitExceeded(format!(
400                "CloudEvents JSON string exceeds {MAX_CLOUDEVENTS_STRING_BYTES} bytes"
401            )));
402        }
403        _ => {}
404    }
405    Ok(())
406}
407
408fn preflight_depth(text: &str) -> Result<()> {
409    let mut depth = 0usize;
410    let mut quoted = false;
411    let mut escaped = false;
412    for byte in text.bytes() {
413        if quoted {
414            if escaped {
415                escaped = false;
416            } else if byte == b'\\' {
417                escaped = true;
418            } else if byte == b'"' {
419                quoted = false;
420            }
421            continue;
422        }
423        match byte {
424            b'"' => quoted = true,
425            b'{' | b'[' => {
426                depth += 1;
427                if depth > MAX_CLOUDEVENTS_DEPTH {
428                    return Err(Error::LimitExceeded(format!(
429                        "CloudEvents JSON nesting exceeds {MAX_CLOUDEVENTS_DEPTH} levels"
430                    )));
431                }
432            }
433            b'}' | b']' => depth = depth.saturating_sub(1),
434            _ => {}
435        }
436    }
437    Ok(())
438}
439
440#[cfg(test)]
441mod tests {
442    use super::*;
443
444    #[test]
445    fn recognizes_cloud_event_object_and_batch() {
446        assert!(looks_like_prefix(
447            br#"{"specversion":"1.0","type":"com.example.created","source":"/demo","id":"1"}"#
448        ));
449        assert!(looks_like_prefix(
450            br#"[{"specversion":"1.0","type":"com.example.created","source":"/demo","id":"1"}]"#
451        ));
452        assert!(!looks_like_prefix(
453            br#"{"type":"thing","source":"/demo","id":"1"}"#
454        ));
455        assert!(!looks_like_prefix(
456            br#"{"version":"https://jsonfeed.org/version/1.1","title":"Feed","items":[{"specversion":"1.0","type":"thing","source":"/demo","id":"1"}]}"#
457        ));
458    }
459
460    #[test]
461    fn summarizes_payload_and_masks_uri_values() {
462        let (table, metadata, warnings) = parse(
463            r#"{"specversion":"1.0","type":"com.example.created","source":"https://example.invalid/events?token=secret&kind=test","id":"evt-1","time":"2024-05-12T00:00:00Z","subject":"item-1","datacontenttype":"application/json","dataschema":"https://schema.invalid/event.json","traceparent":"secret-extension","data":{"message":"very-secret"}}"#,
464        )
465        .unwrap();
466        assert!(metadata.contains("Events: 1"));
467        assert_eq!(table.rows[0][0], "com.example.created");
468        assert!(table.rows[0][2].contains("example.invalid"));
469        assert!(
470            !table
471                .rows
472                .iter()
473                .flatten()
474                .any(|value| value.contains("secret"))
475        );
476        assert!(table.rows[0][2].contains("object/"));
477        assert!(metadata.contains("Extension attributes: 1"));
478        assert!(warnings.iter().any(|warning| warning.contains("payloads")));
479    }
480
481    #[test]
482    fn accepts_json_batch_and_base64_summary() {
483        let (table, metadata, _) = parse(
484            r#"[{"specversion":"1.0","type":"a","source":"/a","id":"1"},{"specversion":"1.0","type":"b","source":"/b","id":"2","data_base64":"c2VjcmV0"}]"#,
485        )
486        .unwrap();
487        assert_eq!(table.rows.len(), 2);
488        assert!(table.rows[1][2].contains("base64 (8 chars)"));
489        assert!(metadata.contains("With data_base64: 1"));
490    }
491
492    #[test]
493    fn rejects_conflicting_payload_encodings() {
494        let result = parse(
495            r#"{"specversion":"1.0","type":"a","source":"/a","id":"1","data":null,"data_base64":"AA=="}"#,
496        );
497        assert!(result.is_err());
498    }
499}
document_svg/document/cloudevents.rs

document_svg/document/
cloudevents.rs
