etdl_parser/
asyncapi.rs

1use crate::ast::{EtlDocument, ExternalRef, MessageRef};
2use crate::jsonptr;
3use serde_json::Value;
4use std::borrow::Cow;
5use std::collections::BTreeMap;
6use std::fs;
7use std::path::{Path, PathBuf};
8
9pub struct AsyncApiRegistry {
10    documents: BTreeMap<String, AsyncApiDocument>,
11}
12
13struct AsyncApiDocument {
14    _path: PathBuf,
15    root: Value,
16}
17
18impl Default for AsyncApiRegistry {
19    fn default() -> Self {
20        Self::new()
21    }
22}
23
24impl AsyncApiRegistry {
25    pub fn new() -> Self {
26        AsyncApiRegistry {
27            documents: BTreeMap::new(),
28        }
29    }
30
31    pub fn load(&mut self, alias: &str, location: &str, base_dir: &Path) -> Result<(), String> {
32        let resolved_path = resolve_location(location, base_dir)?;
33        let content = fs::read_to_string(&resolved_path).map_err(|e| {
34            format!(
35                "cannot read AsyncAPI doc '{}': {}",
36                resolved_path.display(),
37                e
38            )
39        })?;
40
41        let root: Value = if resolved_path.extension().is_some_and(|ext| ext == "json") {
42            serde_json::from_str(&content).map_err(|e| {
43                format!(
44                    "invalid JSON in AsyncAPI doc '{}': {}",
45                    resolved_path.display(),
46                    e
47                )
48            })?
49        } else {
50            serde_yaml::from_str(&content).map_err(|e| {
51                format!(
52                    "invalid YAML in AsyncAPI doc '{}': {}",
53                    resolved_path.display(),
54                    e
55                )
56            })?
57        };
58
59        self.documents.insert(
60            alias.to_string(),
61            AsyncApiDocument {
62                _path: resolved_path,
63                root,
64            },
65        );
66        Ok(())
67    }
68
69    pub fn load_from_content(
70        &mut self,
71        alias: &str,
72        content: &str,
73        is_json: bool,
74    ) -> Result<(), String> {
75        let root: Value = if is_json {
76            serde_json::from_str(content)
77                .map_err(|e| format!("invalid JSON in AsyncAPI doc '{}': {}", alias, e))?
78        } else {
79            serde_yaml::from_str(content)
80                .map_err(|e| format!("invalid YAML in AsyncAPI doc '{}': {}", alias, e))?
81        };
82
83        self.documents.insert(
84            alias.to_string(),
85            AsyncApiDocument {
86                _path: PathBuf::from(alias),
87                root,
88            },
89        );
90        Ok(())
91    }
92
93    pub fn resolve(&self, ext_ref: &ExternalRef) -> Result<&Value, String> {
94        let doc = self.documents.get(&ext_ref.alias).ok_or_else(|| {
95            format!(
96                "import alias '{}' not found in loaded AsyncAPI documents",
97                ext_ref.alias
98            )
99        })?;
100
101        let pointer = &ext_ref.pointer;
102        jsonptr::resolve_json_pointer(&doc.root, pointer).ok_or_else(|| {
103            format!(
104                "JSON Pointer '{}' does not resolve in AsyncAPI document '{}'",
105                pointer, ext_ref.alias
106            )
107        })
108    }
109
110    pub fn resolve_ref(&self, alias: &str, pointer: &str) -> Result<&Value, String> {
111        let doc = self.documents.get(alias).ok_or_else(|| {
112            format!(
113                "import alias '{}' not found in loaded AsyncAPI documents",
114                alias
115            )
116        })?;
117
118        jsonptr::resolve_json_pointer(&doc.root, pointer).ok_or_else(|| {
119            format!(
120                "JSON Pointer '{}' does not resolve in AsyncAPI document '{}'",
121                pointer, alias
122            )
123        })
124    }
125
126    pub fn get_schema_for_path(
127        &self,
128        ext_ref: &ExternalRef,
129        path_segments: &[crate::ecel::PathSegment],
130    ) -> Result<Option<Value>, String> {
131        let message_value = self.resolve(ext_ref)?;
132
133        let payload_schema = message_value
134            .get("payload")
135            .or_else(|| message_value.get("schema"));
136
137        let Some(schema) = payload_schema else {
138            return Ok(None);
139        };
140
141        let resolved = resolve_schema_path(schema, path_segments);
142        Ok(resolved)
143    }
144
145    /// Resolve a Message Reference (Section 5.3.4) to its AsyncAPI Message
146    /// Object-shaped value, regardless of whether it's an External
147    /// Reference (delegates to `resolve`) or an Internal Reference
148    /// (`#/components/messages/<id>`, looked up on `doc` and re-shaped into
149    /// the same `{name, payload, headers}` envelope an External Reference
150    /// would already carry, so callers don't need to branch on the
151    /// reference kind).
152    pub fn resolve_message<'a>(
153        &'a self,
154        doc: &EtlDocument,
155        msg_ref: &MessageRef,
156    ) -> Result<Cow<'a, Value>, String> {
157        match msg_ref {
158            MessageRef::External(ext_ref) => self.resolve(ext_ref).map(Cow::Borrowed),
159            MessageRef::Internal(int_ref) => {
160                let id = internal_message_id(&int_ref.pointer).ok_or_else(|| {
161                    format!(
162                        "internal reference '{}' is not of the form #/components/messages/<id>",
163                        int_ref.pointer
164                    )
165                })?;
166                let message = doc
167                    .components
168                    .as_ref()
169                    .and_then(|c| c.messages.as_ref())
170                    .and_then(|m| m.get(id))
171                    .ok_or_else(|| {
172                        format!(
173                            "internal reference '{}' does not resolve: no components.messages.{}",
174                            int_ref.pointer, id
175                        )
176                    })?;
177                let value = serde_json::to_value(message).map_err(|e| {
178                    format!("cannot convert inline message '{}' to JSON: {}", id, e)
179                })?;
180                Ok(Cow::Owned(value))
181            }
182        }
183    }
184
185    /// Like `get_schema_for_path`, but accepts either kind of Message
186    /// Reference (see `resolve_message`).
187    pub fn get_schema_for_message_ref(
188        &self,
189        doc: &EtlDocument,
190        msg_ref: &MessageRef,
191        path_segments: &[crate::ecel::PathSegment],
192    ) -> Result<Option<Value>, String> {
193        let message_value = self.resolve_message(doc, msg_ref)?;
194
195        let payload_schema = message_value
196            .get("payload")
197            .or_else(|| message_value.get("schema"));
198
199        let Some(schema) = payload_schema else {
200            return Ok(None);
201        };
202
203        Ok(resolve_schema_path(schema, path_segments))
204    }
205}
206
207/// Extracts `<id>` from an internal message-reference pointer of the form
208/// `#/components/messages/<id>`, or `None` if the pointer doesn't match
209/// that shape (e.g. it's a fault-tree `probabilitySource` pointer instead).
210fn internal_message_id(pointer: &str) -> Option<&str> {
211    let id = pointer.strip_prefix("#/components/messages/")?;
212    if id.is_empty() || id.contains('/') {
213        None
214    } else {
215        Some(id)
216    }
217}
218
219fn resolve_schema_path(schema: &Value, segments: &[crate::ecel::PathSegment]) -> Option<Value> {
220    if segments.is_empty() {
221        return Some(schema.clone());
222    }
223
224    let first = &segments[0];
225
226    match first {
227        crate::ecel::PathSegment::Field(name) => {
228            // `message` and `payload` are root markers, not real fields:
229            // `get_schema_for_path` (the only caller) already unwraps the
230            // message envelope down to its payload schema before calling
231            // this function, so `schema` here already *is* what
232            // `message.payload` denotes. Without stripping `payload` too,
233            // every `message.payload.<field>` path (the standard ECEL
234            // root — see `docs/reference/cli.md`/§6.3) tried to resolve a
235            // literal field named `payload` inside the payload schema,
236            // which essentially never exists, so this always returned
237            // `None` -> the caller treated the type as `Unknown` ->
238            // V-204 type-checking silently never fired for any path
239            // operand. `message.headers.<field>` is not fixed by this —
240            // header schema introspection is a separate, already-known,
241            // documented gap (`docs/SPEC_IMPLEMENTATION_MATRIX.md` §6.3).
242            if (name == "message" || name == "payload") && segments.len() > 1 {
243                return resolve_schema_path(schema, &segments[1..]);
244            }
245
246            let field_schema = resolve_field(schema, name)?;
247            if segments.len() == 1 {
248                Some(field_schema.clone())
249            } else {
250                resolve_schema_path(&field_schema, &segments[1..])
251            }
252        }
253        crate::ecel::PathSegment::Wildcard => {
254            let items_schema = resolve_array_items(schema)?;
255            if segments.len() == 1 {
256                Some(items_schema.clone())
257            } else {
258                resolve_schema_path(&items_schema, &segments[1..])
259            }
260        }
261        crate::ecel::PathSegment::Index(_) => {
262            let items_schema = resolve_array_items(schema)?;
263            if segments.len() == 1 {
264                Some(items_schema.clone())
265            } else {
266                resolve_schema_path(&items_schema, &segments[1..])
267            }
268        }
269        crate::ecel::PathSegment::QuotedKey(name) => {
270            let field_schema = resolve_field(schema, name)?;
271            if segments.len() == 1 {
272                Some(field_schema.clone())
273            } else {
274                resolve_schema_path(&field_schema, &segments[1..])
275            }
276        }
277    }
278}
279
280fn resolve_field(schema: &Value, name: &str) -> Option<Value> {
281    if let Some(properties) = schema.get("properties") {
282        if let Some(field) = properties.get(name) {
283            return Some(field.clone());
284        }
285    }
286
287    if let Some(obj) = schema.as_object() {
288        if let Some(field) = obj.get(name) {
289            return Some(field.clone());
290        }
291    }
292
293    None
294}
295
296fn resolve_array_items(schema: &Value) -> Option<Value> {
297    if let Some(items) = schema.get("items") {
298        return Some(items.clone());
299    }
300
301    if let Some(type_val) = schema.get("type") {
302        if type_val.as_str() == Some("array") {
303            if let Some(items) = schema.get("items") {
304                return Some(items.clone());
305            }
306        }
307    }
308
309    None
310}
311
312fn resolve_location(location: &str, base_dir: &Path) -> Result<PathBuf, String> {
313    if location.starts_with("http://") || location.starts_with("https://") {
314        return Err(format!(
315            "remote AsyncAPI imports not supported in this version: '{}'",
316            location
317        ));
318    }
319
320    let path = Path::new(location);
321
322    if path.is_absolute() {
323        // Absolute paths are allowed as-is (caller-provided and trusted).
324        return Ok(path.to_path_buf());
325    }
326
327    // Reject `..` escapes outside the project root (ETDL §12: local imports
328    // MUST NOT escape the project root).
329    if location.split('/').any(|seg| seg == "..") {
330        return Err(format!(
331            "AsyncAPI import '{}' must not contain '..' (path traversal outside the project root is forbidden)",
332            location
333        ));
334    }
335
336    Ok(base_dir.join(path))
337}
338
339#[cfg(test)]
340mod tests {
341    use super::*;
342    use std::path::Path;
343
344    #[test]
345    fn rejects_path_traversal() {
346        let base = Path::new("/proj");
347        assert!(resolve_location("../../etc/passwd", base).is_err());
348        assert!(resolve_location("./../secret.yaml", base).is_err());
349        assert!(resolve_location("a/../b.yaml", base).is_err());
350    }
351
352    #[test]
353    fn accepts_local_and_absolute() {
354        let base = Path::new("/proj");
355        assert_eq!(
356            resolve_location("api.yaml", base).unwrap(),
357            Path::new("/proj/api.yaml")
358        );
359        assert_eq!(
360            resolve_location("/etc/api.yaml", base).unwrap(),
361            Path::new("/etc/api.yaml")
362        );
363    }
364
365    #[test]
366    fn rejects_remote() {
367        let base = Path::new("/proj");
368        assert!(resolve_location("https://example.com/api.yaml", base).is_err());
369    }
370
371    #[test]
372    fn load_from_content_roundtrip() {
373        let mut registry = AsyncApiRegistry::new();
374        let yaml =
375            "asyncapi: '3.0.0'\ninfo:\n  title: t\n  version: '1'\nchannels: {}\ncomponents: {}\n";
376        registry.load_from_content("api", yaml, false).unwrap();
377        let ext = ExternalRef {
378            alias: "api".to_string(),
379            pointer: "/info/title".to_string(),
380        };
381        assert_eq!(registry.resolve(&ext).unwrap(), &serde_json::json!("t"));
382    }
383}
etdl_parser/asyncapi.rs

etdl_parser/
asyncapi.rs
