etdl_parser/
asyncapi.rs

1use crate::ast::ExternalRef;
2use crate::jsonptr;
3use serde_json::Value;
4use std::collections::BTreeMap;
5use std::fs;
6use std::path::{Path, PathBuf};
7
8pub struct AsyncApiRegistry {
9    documents: BTreeMap<String, AsyncApiDocument>,
10}
11
12struct AsyncApiDocument {
13    _path: PathBuf,
14    root: Value,
15}
16
17impl Default for AsyncApiRegistry {
18    fn default() -> Self {
19        Self::new()
20    }
21}
22
23impl AsyncApiRegistry {
24    pub fn new() -> Self {
25        AsyncApiRegistry {
26            documents: BTreeMap::new(),
27        }
28    }
29
30    pub fn load(&mut self, alias: &str, location: &str, base_dir: &Path) -> Result<(), String> {
31        let resolved_path = resolve_location(location, base_dir)?;
32        let content = fs::read_to_string(&resolved_path).map_err(|e| {
33            format!(
34                "cannot read AsyncAPI doc '{}': {}",
35                resolved_path.display(),
36                e
37            )
38        })?;
39
40        let root: Value = if resolved_path.extension().is_some_and(|ext| ext == "json") {
41            serde_json::from_str(&content).map_err(|e| {
42                format!(
43                    "invalid JSON in AsyncAPI doc '{}': {}",
44                    resolved_path.display(),
45                    e
46                )
47            })?
48        } else {
49            serde_yaml::from_str(&content).map_err(|e| {
50                format!(
51                    "invalid YAML in AsyncAPI doc '{}': {}",
52                    resolved_path.display(),
53                    e
54                )
55            })?
56        };
57
58        self.documents.insert(
59            alias.to_string(),
60            AsyncApiDocument {
61                _path: resolved_path,
62                root,
63            },
64        );
65        Ok(())
66    }
67
68    pub fn load_from_content(
69        &mut self,
70        alias: &str,
71        content: &str,
72        is_json: bool,
73    ) -> Result<(), String> {
74        let root: Value = if is_json {
75            serde_json::from_str(content)
76                .map_err(|e| format!("invalid JSON in AsyncAPI doc '{}': {}", alias, e))?
77        } else {
78            serde_yaml::from_str(content)
79                .map_err(|e| format!("invalid YAML in AsyncAPI doc '{}': {}", alias, e))?
80        };
81
82        self.documents.insert(
83            alias.to_string(),
84            AsyncApiDocument {
85                _path: PathBuf::from(alias),
86                root,
87            },
88        );
89        Ok(())
90    }
91
92    pub fn resolve(&self, ext_ref: &ExternalRef) -> Result<&Value, String> {
93        let doc = self.documents.get(&ext_ref.alias).ok_or_else(|| {
94            format!(
95                "import alias '{}' not found in loaded AsyncAPI documents",
96                ext_ref.alias
97            )
98        })?;
99
100        let pointer = &ext_ref.pointer;
101        jsonptr::resolve_json_pointer(&doc.root, pointer).ok_or_else(|| {
102            format!(
103                "JSON Pointer '{}' does not resolve in AsyncAPI document '{}'",
104                pointer, ext_ref.alias
105            )
106        })
107    }
108
109    pub fn resolve_ref(&self, alias: &str, pointer: &str) -> Result<&Value, String> {
110        let doc = self.documents.get(alias).ok_or_else(|| {
111            format!(
112                "import alias '{}' not found in loaded AsyncAPI documents",
113                alias
114            )
115        })?;
116
117        jsonptr::resolve_json_pointer(&doc.root, pointer).ok_or_else(|| {
118            format!(
119                "JSON Pointer '{}' does not resolve in AsyncAPI document '{}'",
120                pointer, alias
121            )
122        })
123    }
124
125    pub fn get_schema_for_path(
126        &self,
127        ext_ref: &ExternalRef,
128        path_segments: &[crate::ecel::PathSegment],
129    ) -> Result<Option<Value>, String> {
130        let message_value = self.resolve(ext_ref)?;
131
132        let payload_schema = message_value
133            .get("payload")
134            .or_else(|| message_value.get("schema"));
135
136        let Some(schema) = payload_schema else {
137            return Ok(None);
138        };
139
140        let resolved = resolve_schema_path(schema, path_segments);
141        Ok(resolved)
142    }
143}
144
145fn resolve_schema_path(schema: &Value, segments: &[crate::ecel::PathSegment]) -> Option<Value> {
146    if segments.is_empty() {
147        return Some(schema.clone());
148    }
149
150    let first = &segments[0];
151
152    match first {
153        crate::ecel::PathSegment::Field(name) => {
154            if name == "message" && segments.len() > 1 {
155                return resolve_schema_path(schema, &segments[1..]);
156            }
157
158            let field_schema = resolve_field(schema, name)?;
159            if segments.len() == 1 {
160                Some(field_schema.clone())
161            } else {
162                resolve_schema_path(&field_schema, &segments[1..])
163            }
164        }
165        crate::ecel::PathSegment::Wildcard => {
166            let items_schema = resolve_array_items(schema)?;
167            if segments.len() == 1 {
168                Some(items_schema.clone())
169            } else {
170                resolve_schema_path(&items_schema, &segments[1..])
171            }
172        }
173        crate::ecel::PathSegment::Index(_) => {
174            let items_schema = resolve_array_items(schema)?;
175            if segments.len() == 1 {
176                Some(items_schema.clone())
177            } else {
178                resolve_schema_path(&items_schema, &segments[1..])
179            }
180        }
181        crate::ecel::PathSegment::QuotedKey(name) => {
182            let field_schema = resolve_field(schema, name)?;
183            if segments.len() == 1 {
184                Some(field_schema.clone())
185            } else {
186                resolve_schema_path(&field_schema, &segments[1..])
187            }
188        }
189    }
190}
191
192fn resolve_field(schema: &Value, name: &str) -> Option<Value> {
193    if let Some(properties) = schema.get("properties") {
194        if let Some(field) = properties.get(name) {
195            return Some(field.clone());
196        }
197    }
198
199    if let Some(obj) = schema.as_object() {
200        if let Some(field) = obj.get(name) {
201            return Some(field.clone());
202        }
203    }
204
205    None
206}
207
208fn resolve_array_items(schema: &Value) -> Option<Value> {
209    if let Some(items) = schema.get("items") {
210        return Some(items.clone());
211    }
212
213    if let Some(type_val) = schema.get("type") {
214        if type_val.as_str() == Some("array") {
215            if let Some(items) = schema.get("items") {
216                return Some(items.clone());
217            }
218        }
219    }
220
221    None
222}
223
224fn resolve_location(location: &str, base_dir: &Path) -> Result<PathBuf, String> {
225    if location.starts_with("http://") || location.starts_with("https://") {
226        return Err(format!(
227            "remote AsyncAPI imports not supported in this version: '{}'",
228            location
229        ));
230    }
231
232    let path = Path::new(location);
233
234    if path.is_absolute() {
235        // Absolute paths are allowed as-is (caller-provided and trusted).
236        return Ok(path.to_path_buf());
237    }
238
239    // Reject `..` escapes outside the project root (ETDL §12: local imports
240    // MUST NOT escape the project root).
241    if location.split('/').any(|seg| seg == "..") {
242        return Err(format!(
243            "AsyncAPI import '{}' must not contain '..' (path traversal outside the project root is forbidden)",
244            location
245        ));
246    }
247
248    Ok(base_dir.join(path))
249}
250
251#[cfg(test)]
252mod tests {
253    use super::*;
254    use std::path::Path;
255
256    #[test]
257    fn rejects_path_traversal() {
258        let base = Path::new("/proj");
259        assert!(resolve_location("../../etc/passwd", base).is_err());
260        assert!(resolve_location("./../secret.yaml", base).is_err());
261        assert!(resolve_location("a/../b.yaml", base).is_err());
262    }
263
264    #[test]
265    fn accepts_local_and_absolute() {
266        let base = Path::new("/proj");
267        assert_eq!(
268            resolve_location("api.yaml", base).unwrap(),
269            Path::new("/proj/api.yaml")
270        );
271        assert_eq!(
272            resolve_location("/etc/api.yaml", base).unwrap(),
273            Path::new("/etc/api.yaml")
274        );
275    }
276
277    #[test]
278    fn rejects_remote() {
279        let base = Path::new("/proj");
280        assert!(resolve_location("https://example.com/api.yaml", base).is_err());
281    }
282
283    #[test]
284    fn load_from_content_roundtrip() {
285        let mut registry = AsyncApiRegistry::new();
286        let yaml =
287            "asyncapi: '3.0.0'\ninfo:\n  title: t\n  version: '1'\nchannels: {}\ncomponents: {}\n";
288        registry.load_from_content("api", yaml, false).unwrap();
289        let ext = ExternalRef {
290            alias: "api".to_string(),
291            pointer: "/info/title".to_string(),
292        };
293        assert_eq!(registry.resolve(&ext).unwrap(), &serde_json::json!("t"));
294    }
295}
etdl_parser/asyncapi.rs

etdl_parser/
asyncapi.rs
