av_bridge/
manifest.rs

1//! Declarative topic-schema manifest + provisioning (the Module F
2//! portability contract: "a customer stands up an identical bridge in a new
3//! region or air-gapped enclave from that manifest alone").
4
5use serde::{Deserialize, Serialize};
6#[cfg(any(feature = "nats", feature = "kafka"))]
7use std::collections::HashMap;
8use std::path::{Path, PathBuf};
9
10/// Manifest format version (evolution surface).
11pub const MANIFEST_VERSION: u32 = 1;
12
13/// Retention policy for one topic.
14#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
15#[serde(deny_unknown_fields)]
16pub struct RetentionSpec {
17    /// Hot retention window in hours (default 720 = 30 days, per the brief).
18    pub hot_hours: u32,
19    /// Optional cold-tier export directory/bucket URI (customer-owned storage).
20    #[serde(default, skip_serializing_if = "Option::is_none")]
21    pub cold_uri: Option<String>,
22}
23
24impl Default for RetentionSpec {
25    fn default() -> Self {
26        Self {
27            hot_hours: 720,
28            cold_uri: None,
29        }
30    }
31}
32
33/// One topic declaration.
34#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
35#[serde(deny_unknown_fields)]
36pub struct TopicSpec {
37    /// Topic name (`agent.tool_call`, …).
38    pub name: String,
39    /// Partition count (partitioned by `ai_agent.instance_uid`).
40    pub partitions: u32,
41    /// Retention policy.
42    #[serde(default)]
43    pub retention: RetentionSpec,
44    /// JSON Schema reference events on this topic must satisfy.
45    #[serde(default, skip_serializing_if = "Option::is_none")]
46    pub schema_ref: Option<String>,
47}
48
49/// The Bridge manifest.
50#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
51#[serde(deny_unknown_fields)]
52pub struct BridgeManifest {
53    /// Manifest format version.
54    pub manifest_version: u32,
55    /// Deployment name (region/enclave label).
56    pub name: String,
57    /// Broker replication factor for managed Kafka/NATS deployments.
58    #[serde(default = "default_replication_factor")]
59    pub replication_factor: u32,
60    /// Topic declarations.
61    pub topics: Vec<TopicSpec>,
62}
63
64fn default_replication_factor() -> u32 {
65    1
66}
67
68/// Manifest validation errors.
69#[derive(Debug, thiserror::Error, PartialEq, Eq)]
70#[non_exhaustive]
71pub enum ManifestError {
72    /// Unsupported version.
73    #[error("unsupported manifest_version {0} (this build supports {MANIFEST_VERSION})")]
74    Version(u32),
75    /// YAML/JSON parse failure.
76    #[error("manifest parse: {0}")]
77    Parse(String),
78    /// Structural problem.
79    #[error("manifest invalid: {0}")]
80    Invalid(String),
81}
82
83impl BridgeManifest {
84    /// The default manifest covering every OCSF event class topic.
85    pub fn default_for(name: &str) -> Self {
86        Self {
87            manifest_version: MANIFEST_VERSION,
88            name: name.to_owned(),
89            replication_factor: default_replication_factor(),
90            topics: av_events::EventClass::all()
91                .iter()
92                .map(|c| TopicSpec {
93                    name: c.topic().to_owned(),
94                    partitions: 8,
95                    retention: RetentionSpec::default(),
96                    schema_ref: Some("schemas/ocsf-agent-event.schema.json".to_owned()),
97                })
98                .collect(),
99        }
100    }
101
102    /// Parse from YAML.
103    ///
104    /// Enforces a 256 KiB input cap and rejects any document containing
105    /// YAML anchors (`&`) or aliases (`*`). Together these close the
106    /// billion-laughs attack surface: `serde_yaml` 0.9 expands aliases
107    /// eagerly with no depth or expansion cap, so a maliciously crafted
108    /// manifest of a few KiB would OOM the process at parse time.
109    /// Legitimate bridge manifests never need aliases — reject rather
110    /// than trying to bound expansion, which would require a fork.
111    pub fn from_yaml(yaml: &str) -> Result<Self, ManifestError> {
112        const MAX_YAML_BYTES: usize = 256 * 1024;
113        if yaml.len() > MAX_YAML_BYTES {
114            return Err(ManifestError::Parse(format!(
115                "manifest is {} bytes, exceeds cap of {MAX_YAML_BYTES}",
116                yaml.len()
117            )));
118        }
119        // Cheap syntactic scan for YAML aliases. Legitimate anchors/
120        // aliases would need explicit `&name`/`*name` syntax; a naive
121        // string search catches every case that serde_yaml would
122        // actually expand. False positives on strings that happen to
123        // contain `&` or `*` are reduced (not eliminated — the scan is
124        // not YAML-context-aware, so a quoted string like `"a &name"`
125        // still trips it; failing closed is acceptable for manifests)
126        // by checking that the char
127        // precedes a valid anchor name character (alphanumeric or `_`).
128        for (marker, kind) in [('&', "anchor"), ('*', "alias")] {
129            let mut chars = yaml.char_indices().peekable();
130            while let Some((_, ch)) = chars.next() {
131                if ch != marker {
132                    continue;
133                }
134                if let Some(&(_, next)) = chars.peek() {
135                    if next.is_ascii_alphanumeric() || next == '_' {
136                        return Err(ManifestError::Parse(format!(
137                            "manifest contains a YAML {kind} ('{marker}<name>'); \
138                             AgentVisor AI refuses anchor/alias syntax to close the \
139                             billion-laughs attack surface (serde_yaml expands aliases \
140                             with no cap). Rewrite the document without &/* references."
141                        )));
142                    }
143                }
144            }
145        }
146        let m: Self = serde_yaml::from_str(yaml).map_err(|e| ManifestError::Parse(e.to_string()))?;
147        m.validate()?;
148        Ok(m)
149    }
150
151    /// Serialize to YAML.
152    pub fn to_yaml(&self) -> Result<String, ManifestError> {
153        serde_yaml::to_string(self).map_err(|e| ManifestError::Parse(e.to_string()))
154    }
155
156    /// Validate structural invariants.
157    pub fn validate(&self) -> Result<(), ManifestError> {
158        if self.manifest_version != MANIFEST_VERSION {
159            return Err(ManifestError::Version(self.manifest_version));
160        }
161        if self.name.is_empty() {
162            return Err(ManifestError::Invalid("name is empty".into()));
163        }
164        if self.topics.is_empty() {
165            return Err(ManifestError::Invalid("no topics declared".into()));
166        }
167        if !(1..=5).contains(&self.replication_factor) {
168            return Err(ManifestError::Invalid(
169                "replication_factor must be between 1 and 5".to_owned(),
170            ));
171        }
172        let mut names: Vec<&str> = self.topics.iter().map(|t| t.name.as_str()).collect();
173        names.sort_unstable();
174        let before = names.len();
175        names.dedup();
176        if names.len() != before {
177            return Err(ManifestError::Invalid("duplicate topic names".into()));
178        }
179        for t in &self.topics {
180            if t.name.is_empty()
181                || !t
182                    .name
183                    .bytes()
184                    .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b'-'))
185                || t.name == "."
186                || t.name == ".."
187            {
188                return Err(ManifestError::Invalid(format!("unsafe topic name {:?}", t.name)));
189            }
190            if t.partitions == 0 {
191                return Err(ManifestError::Invalid(format!(
192                    "topic {:?} has 0 partitions",
193                    t.name
194                )));
195            }
196            // Round-27 F5: cap partitions and hot_hours. `partitions` is
197            // u32, and without a cap `partitions: 4294967295` would
198            // pass validate() and drive the embedded broker to create
199            // 4 billion partition directories on startup — guaranteed
200            // inode/OOM exhaustion. 1024 matches the practical Kafka
201            // per-topic sanity cap. 87_600 hours is 10 years —
202            // realistic ceiling for a long-lived audit trail.
203            const MAX_PARTITIONS: u32 = 1024;
204            const MAX_HOT_HOURS: u32 = 24 * 365 * 10;
205            if t.partitions > MAX_PARTITIONS {
206                return Err(ManifestError::Invalid(format!(
207                    "topic {:?} partitions {} exceeds cap of {MAX_PARTITIONS}",
208                    t.name, t.partitions
209                )));
210            }
211            if t.retention.hot_hours == 0 {
212                return Err(ManifestError::Invalid(format!(
213                    "topic {:?} has 0h retention",
214                    t.name
215                )));
216            }
217            if t.retention.hot_hours > MAX_HOT_HOURS {
218                return Err(ManifestError::Invalid(format!(
219                    "topic {:?} hot_hours {} exceeds cap of {MAX_HOT_HOURS} (10 years)",
220                    t.name, t.retention.hot_hours
221                )));
222            }
223            if let Some(reference) = &t.schema_ref {
224                let path = Path::new(reference);
225                if path.is_absolute()
226                    || path
227                        .components()
228                        .any(|component| !matches!(component, std::path::Component::Normal(_)))
229                {
230                    return Err(ManifestError::Invalid(format!("unsafe schema_ref {reference:?}")));
231                }
232            }
233        }
234        Ok(())
235    }
236}
237
238#[cfg(any(feature = "nats", feature = "kafka"))]
239pub(crate) fn compile_topic_validators(
240    manifest: &BridgeManifest,
241) -> Result<HashMap<String, jsonschema::Validator>, crate::BusError> {
242    let mut validators = HashMap::new();
243    for topic in &manifest.topics {
244        let Some(reference) = &topic.schema_ref else {
245            continue;
246        };
247        let schema = schema_document(reference)?;
248        let validator = jsonschema::validator_for(&schema)
249            .map_err(|error| crate::BusError::Backend(format!("invalid schema {reference:?}: {error}")))?;
250        validators.insert(topic.name.clone(), validator);
251    }
252    Ok(validators)
253}
254
255#[cfg(any(feature = "nats", feature = "kafka"))]
256pub(crate) fn validate_topic_event(
257    validators: &HashMap<String, jsonschema::Validator>,
258    topic: &str,
259    value: &serde_json::Value,
260) -> Result<(), crate::BusError> {
261    let Some(validator) = validators.get(topic) else {
262        return Ok(());
263    };
264    let errors: Vec<String> = validator
265        .iter_errors(value)
266        .take(3)
267        .map(|error| error.to_string())
268        .collect();
269    if errors.is_empty() {
270        Ok(())
271    } else {
272        Err(crate::BusError::Backend(format!(
273            "event rejected by schema for topic {topic:?}: {}",
274            errors.join("; ")
275        )))
276    }
277}
278
279pub(crate) fn schema_document(reference: &str) -> Result<serde_json::Value, crate::BusError> {
280    if reference == "schemas/ocsf-agent-event.schema.json" {
281        return serde_json::from_str(include_str!("../../../schemas/ocsf-agent-event.schema.json"))
282            .map_err(crate::BusError::from);
283    }
284    let direct = PathBuf::from(reference);
285    if direct.exists() {
286        return serde_json::from_slice(&std::fs::read(direct)?).map_err(crate::BusError::from);
287    }
288    let workspace = Path::new(env!("CARGO_MANIFEST_DIR"))
289        .join("../..")
290        .join(reference);
291    if workspace.exists() {
292        serde_json::from_slice(&std::fs::read(workspace)?).map_err(crate::BusError::from)
293    } else {
294        Err(crate::BusError::Backend(format!(
295            "schema reference {reference:?} could not be resolved"
296        )))
297    }
298}
299
300#[cfg(test)]
301mod tests {
302    #![allow(
303        clippy::unwrap_used,
304        clippy::expect_used,
305        clippy::panic,
306        clippy::indexing_slicing
307    )]
308
309    use super::*;
310
311    #[test]
312    fn default_manifest_covers_all_event_classes() {
313        let m = BridgeManifest::default_for("us-east-lab");
314        m.validate().unwrap();
315        assert_eq!(m.topics.len(), av_events::EventClass::all().len());
316        assert!(m.topics.iter().any(|t| t.name == "agent.tool_call"));
317        assert!(m.topics.iter().any(|t| t.name == "agent.receipt"));
318        assert_eq!(m.topics[0].retention.hot_hours, 720, "default 30 days per brief");
319    }
320
321    #[test]
322    fn yaml_roundtrip() {
323        let m = BridgeManifest::default_for("enclave-1");
324        let yaml = m.to_yaml().unwrap();
325        let back = BridgeManifest::from_yaml(&yaml).unwrap();
326        assert_eq!(m, back);
327    }
328
329    #[test]
330    fn rejects_bad_manifests() {
331        let mut m = BridgeManifest::default_for("x");
332        m.manifest_version = 99;
333        assert_eq!(m.validate(), Err(ManifestError::Version(99)));
334
335        let mut m = BridgeManifest::default_for("x");
336        m.topics[0].partitions = 0;
337        assert!(matches!(m.validate(), Err(ManifestError::Invalid(_))));
338
339        let mut m = BridgeManifest::default_for("x");
340        m.topics[0].name = "../escape".to_owned();
341        assert!(matches!(m.validate(), Err(ManifestError::Invalid(_))));
342
343        let mut m = BridgeManifest::default_for("x");
344        m.topics[0].schema_ref = Some("../outside.json".to_owned());
345        assert!(matches!(m.validate(), Err(ManifestError::Invalid(_))));
346
347        let mut m = BridgeManifest::default_for("x");
348        let dup = m.topics[0].clone();
349        m.topics.push(dup);
350        assert!(matches!(m.validate(), Err(ManifestError::Invalid(_))));
351
352        let m = BridgeManifest {
353            manifest_version: MANIFEST_VERSION,
354            name: String::new(),
355            replication_factor: 1,
356            topics: vec![],
357        };
358        assert!(m.validate().is_err());
359
360        // Round-27 F5: upper caps on partitions and hot_hours. Absurd
361        // values (`u32::MAX`) used to pass validate() and would drive
362        // EmbeddedBroker::provision to create 4 billion partition
363        // directories at startup — guaranteed inode/OOM exhaustion.
364        let mut m = BridgeManifest::default_for("x");
365        m.topics[0].partitions = u32::MAX;
366        assert!(matches!(m.validate(), Err(ManifestError::Invalid(_))));
367        let mut m = BridgeManifest::default_for("x");
368        m.topics[0].retention.hot_hours = u32::MAX;
369        assert!(matches!(m.validate(), Err(ManifestError::Invalid(_))));
370    }
371
372    /// Round-27 F5: `deny_unknown_fields` on `BridgeManifest`,
373    /// `TopicSpec`, and `RetentionSpec` catches config-file typos
374    /// that would otherwise silently disable features. `cold_url`
375    /// (misspelling `cold_uri`) or `hot_days` (misspelling
376    /// `hot_hours`) used to pass `avctl manifest-validate` cleanly
377    /// and cold export or retention would silently do nothing.
378    #[test]
379    fn unknown_fields_are_rejected_at_parse_time() {
380        let with_top_level_typo = r"
381manifest_version: 1
382name: probe
383topocs: []
384";
385        assert!(matches!(
386            BridgeManifest::from_yaml(with_top_level_typo),
387            Err(ManifestError::Parse(_))
388        ));
389        let with_topic_typo = r"
390manifest_version: 1
391name: probe
392topics:
393  - name: a
394    partitions: 1
395    schema_reff: something
396";
397        assert!(matches!(
398            BridgeManifest::from_yaml(with_topic_typo),
399            Err(ManifestError::Parse(_))
400        ));
401        let with_retention_typo = r"
402manifest_version: 1
403name: probe
404topics:
405  - name: a
406    partitions: 1
407    retention:
408      hot_horus: 720
409";
410        assert!(matches!(
411            BridgeManifest::from_yaml(with_retention_typo),
412            Err(ManifestError::Parse(_))
413        ));
414    }
415
416    #[test]
417    fn malformed_yaml_is_a_parse_error() {
418        assert!(matches!(
419            BridgeManifest::from_yaml(":\n  - not: [valid"),
420            Err(ManifestError::Parse(_))
421        ));
422    }
423
424    #[test]
425    fn default_manifest_matches_shipped_json_schema() {
426        let schema: serde_json::Value =
427            serde_json::from_str(include_str!("../../../schemas/bridge-manifest.schema.json")).unwrap();
428        let validator = jsonschema::validator_for(&schema).unwrap();
429        let value = serde_json::to_value(BridgeManifest::default_for("schema-test")).unwrap();
430        let errors: Vec<_> = validator.iter_errors(&value).collect();
431        assert!(errors.is_empty(), "{errors:?}");
432    }
433}
434
435#[cfg(test)]
436mod mutation_boundary_tests {
437    #![allow(
438        clippy::unwrap_used,
439        clippy::expect_used,
440        clippy::panic,
441        clippy::indexing_slicing
442    )]
443
444    use super::*;
445
446    /// Mutation-run hardening (round 14): the alias scan's `||` -> `&&`
447    /// survivor would only refuse markers followed by BOTH alphanumeric
448    /// and `_` (impossible), silently re-opening the YAML anchor-bomb
449    /// vector for `_`-named anchors. Pin each marker/name-class combo,
450    /// and keep innocuous `&`/`*` usage legal.
451    #[test]
452    fn yaml_anchor_and_alias_markers_are_refused_per_name_class() {
453        let base = BridgeManifest::default_for("anchors").to_yaml().unwrap();
454        for snippet in ["&a1", "&_x", "*a1", "*_x"] {
455            let hostile = format!("{base}# {snippet}\n");
456            assert!(
457                BridgeManifest::from_yaml(&hostile).is_err(),
458                "{snippet} must be refused"
459            );
460        }
461        // A trailing bare marker (no name character) stays legal.
462        let benign = format!("{base}# tail & done * \n");
463        BridgeManifest::from_yaml(&benign).unwrap();
464    }
465
466    /// The 256 KiB manifest cap must be exact: at-cap parses, one byte
467    /// past is refused with the cap message (not a YAML error).
468    #[test]
469    fn manifest_size_cap_is_exact() {
470        let base = BridgeManifest::default_for("size-cap").to_yaml().unwrap();
471        let cap = 256 * 1024;
472        let pad_to = |len: usize| {
473            let mut s = base.clone();
474            s.push('#');
475            while s.len() < len {
476                s.push('x');
477            }
478            s
479        };
480        BridgeManifest::from_yaml(&pad_to(cap)).unwrap();
481        let over = BridgeManifest::from_yaml(&pad_to(cap + 1));
482        assert!(
483            matches!(over, Err(ManifestError::Parse(ref m)) if m.contains("exceeds cap")),
484            "one past the cap must refuse, got {over:?}"
485        );
486    }
487
488    /// Path-traversal topic names: `.` and `..` are refused explicitly
489    /// (they pass the character filter), and the partition/hot-hours caps
490    /// hold exactly at their boundaries.
491    #[test]
492    fn topic_name_dots_and_numeric_caps_are_exact() {
493        let mut m = BridgeManifest::default_for("dots");
494        for name in [".", ".."] {
495            m.topics[0].name = name.to_owned();
496            assert!(m.validate().is_err(), "topic name {name:?} must be refused");
497        }
498        m.topics[0].name = "agent.ok".to_owned();
499        m.topics[0].partitions = 1024;
500        m.topics[0].retention.hot_hours = 24 * 365 * 10;
501        m.validate().unwrap();
502        m.topics[0].partitions = 1025;
503        assert!(m.validate().is_err(), "1025 partitions past cap");
504        m.topics[0].partitions = 1;
505        m.topics[0].retention.hot_hours = 24 * 365 * 10 + 1;
506        assert!(m.validate().is_err(), "hot_hours past the 10y cap");
507    }
508
509    /// `validate_topic_event -> Ok(())` survived: with schema_refs set,
510    /// a compile+validate round-trip must actually reject a nonconforming
511    /// event (this is the schema-enforcement path the Kafka and NATS
512    /// connectors share).
513    #[test]
514    #[cfg(any(feature = "nats", feature = "kafka"))]
515    fn compiled_topic_validators_reject_nonconforming_events() {
516        let mut m = BridgeManifest::default_for("schemas");
517        // Give one topic an inline-file schema requiring an object.
518        let dir = tempfile::tempdir().unwrap();
519        let schema_path = dir.path().join("strict.json");
520        std::fs::write(
521            &schema_path,
522            br#"{"type":"object","required":["metadata"],"properties":{"metadata":{"type":"object"}}}"#,
523        )
524        .unwrap();
525        for t in &mut m.topics {
526            t.schema_ref = None;
527        }
528        m.topics[0].schema_ref = Some(schema_path.to_string_lossy().into_owned());
529        let topic = m.topics[0].name.clone();
530        let validators = compile_topic_validators(&m).unwrap();
531        validate_topic_event(&validators, &topic, &serde_json::json!({"metadata": {}})).unwrap();
532        let refused = validate_topic_event(&validators, &topic, &serde_json::json!({"not": "conforming"}));
533        assert!(refused.is_err(), "nonconforming event must be refused");
534    }
535}
av_bridge/manifest.rs

av_bridge/
manifest.rs
