fix: add input validation and CLI consistency across all nodes (#6)

Audit all nodes against CLAUDE.md checklist and fix compliance gaps:

- system-telemetry: add topic name validation and rate_hz bounds checking
- network-monitor: add topic/numeric validation, endpoint default, build task
- openmeteo: add config bounds validation, fix Header scope field, scope
  topics with bubbaloop/{scope}/{machine}/ prefix, rename -z to -e flag,
  add HTTP client timeout
- inference: add -c/-e CLI flags, config.yaml, scoped subscribe topic

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: edgarriba

inference/Cargo.toml

@@ -15,6 +15,10 @@ env_logger = "0.11"
 log = "0.4"
 ctrlc = "3.4"
 hostname = "0.4"
+argh = "0.1"
+serde = { version = "1.0", features = ["derive"] }
+serde_yaml = "0.9"
+regex-lite = "0.1"
 bubbaloop-schemas = { git = "https://github.com/kornia/bubbaloop.git", branch = "main" }
 
 [[bin]]

inference/config.yaml

@@ -0,0 +1,5 @@
+# Instance parameters for inference.
+# Run: ./target/release/inference_node -c config.yaml
+
+# Topic pattern to subscribe to (wildcard for all cameras)
+subscribe_topic: camera/*/raw_shm

inference/node.yaml

@@ -4,4 +4,4 @@ type: rust
 description: Inference node for camera processing
 author: Bubbaloop Team
 build: pixi run build
-command: ./target/release/inference_node
+command: ./target/release/inference_node -c config.yaml

inference/src/bin/inference_node.rs

@@ -5,10 +5,63 @@
 //!
 //! Based on: https://github.com/eclipse-zenoh/zenoh/blob/main/examples/examples/z_sub_shm.rs
 
+use argh::FromArgs;
 use bubbaloop_schemas::RawImage;
 use prost::Message;
+use serde::Deserialize;
+use std::path::{Path, PathBuf};
 use zenoh::Wait;
 
+/// Inference node configuration
+#[derive(Debug, Clone, Deserialize)]
+struct Config {
+    /// Topic pattern to subscribe to
+    subscribe_topic: String,
+}
+
+impl Default for Config {
+    fn default() -> Self {
+        Self {
+            subscribe_topic: "camera/*/raw_shm".to_string(),
+        }
+    }
+}
+
+/// Inference node for camera stream processing (SHM subscriber)
+#[derive(FromArgs)]
+struct Args {
+    /// path to configuration file
+    #[argh(option, short = 'c', default = "default_config_path()")]
+    config: PathBuf,
+
+    /// zenoh endpoint to connect to
+    #[argh(option, short = 'e', default = "default_endpoint()")]
+    endpoint: String,
+}
+
+fn default_config_path() -> PathBuf {
+    PathBuf::from("config.yaml")
+}
+
+fn default_endpoint() -> String {
+    String::from("tcp/127.0.0.1:7447")
+}
+
+fn load_config(path: &Path) -> Config {
+    if path.exists() {
+        match std::fs::read_to_string(path) {
+            Ok(content) => match serde_yaml::from_str(&content) {
+                Ok(config) => return config,
+                Err(e) => log::warn!("Failed to parse config file: {}, using defaults", e),
+            },
+            Err(e) => log::warn!("Failed to read config file: {}, using defaults", e),
+        }
+    } else {
+        log::warn!("Config file not found: {:?}, using defaults", path);
+    }
+    Config::default()
+}
+
 /// Compute mean and standard deviation of pixel values
 fn compute_image_stats(data: &[u8]) -> (f64, f64) {
     if data.is_empty() {
@@ -30,8 +83,22 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     zenoh::init_log_from_env_or("error");
     env_logger::init();
 
+    let args: Args = argh::from_env();
+
     log::info!("Starting inference node (SHM subscriber)...");
 
+    // Load and validate config
+    let config = load_config(&args.config);
+
+    let topic_re = regex_lite::Regex::new(r"^[a-zA-Z0-9/_\-\.\*]+$").unwrap();
+    if !topic_re.is_match(&config.subscribe_topic) {
+        log::error!(
+            "subscribe_topic '{}' contains invalid characters",
+            config.subscribe_topic
+        );
+        std::process::exit(1);
+    }
+
     // Read scope/machine env vars for health heartbeat
     let scope = std::env::var("BUBBALOOP_SCOPE").unwrap_or_else(|_| "local".to_string());
     let machine_id = std::env::var("BUBBALOOP_MACHINE_ID").unwrap_or_else(|_| {
@@ -54,26 +121,30 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     }
 
     // Create Zenoh session with SHM enabled
-    let mut config = zenoh::Config::default();
-    config.insert_json5("transport/shared_memory/enabled", "true")?;
-    if let Ok(endpoint) = std::env::var("ZENOH_ENDPOINT") {
-        config.insert_json5("connect/endpoints", &format!(r#"["{}"]"#, endpoint))?;
-    }
+    let endpoint = std::env::var("ZENOH_ENDPOINT").unwrap_or(args.endpoint);
+    log::info!("Connecting to Zenoh at: {}", endpoint);
+
+    let mut zenoh_config = zenoh::Config::default();
+    zenoh_config.insert_json5("transport/shared_memory/enabled", "true")?;
+    zenoh_config.insert_json5("connect/endpoints", &format!(r#"["{}"]"#, endpoint))?;
 
-    let session = zenoh::open(config).wait()?;
+    let session = zenoh::open(zenoh_config).wait()?;
 
     // Create health heartbeat publisher
     let health_topic = format!("bubbaloop/{}/{}/health/inference", scope, machine_id);
     let health_publisher = session.declare_publisher(health_topic.clone()).await?;
     log::info!("Health heartbeat topic: {}", health_topic);
 
-    // Subscribe to all camera raw_shm topics using wildcard
-    let topic = "camera/*/raw_shm";
-    let subscriber = session.declare_subscriber(topic).wait()?;
+    // Subscribe using scoped topic
+    let full_topic = format!(
+        "bubbaloop/{}/{}/{}",
+        scope, machine_id, config.subscribe_topic
+    );
+    let subscriber = session.declare_subscriber(&full_topic).wait()?;
 
     log::info!(
         "Inference node subscribed to '{}', waiting for SHM images...",
-        topic
+        full_topic
     );
 
     let mut frame_count = 0u64;

network-monitor/main.py

@@ -12,10 +12,15 @@
 from datetime import datetime, timezone
 from pathlib import Path
 
+import re
+
 import requests
 import yaml
 import zenoh
 
+# Topic name validation pattern
+TOPIC_RE = re.compile(r"^[a-zA-Z0-9/_\-\.]+$")
+
 # Configure logging
 logging.basicConfig(
     level=logging.INFO,
@@ -53,6 +58,20 @@ def __init__(self, config_path: Path, endpoint: str | None = None):
                 "checks": [],
             }
 
+        # Validate config
+        topic = self.config.get("publish_topic", "")
+        if not TOPIC_RE.match(topic):
+            raise ValueError(
+                f"publish_topic '{topic}' contains invalid characters "
+                f"(must match [a-zA-Z0-9/_\\-\\.]+)"
+            )
+        rate_hz = self.config.get("rate_hz", 0.1)
+        if not (0.01 <= rate_hz <= 1000.0):
+            raise ValueError(f"rate_hz {rate_hz} out of range (0.01-1000.0)")
+        timeout_secs = self.config.get("timeout_secs", 5)
+        if not (1 <= timeout_secs <= 300):
+            raise ValueError(f"timeout_secs {timeout_secs} out of range (1-300)")
+
         # Resolve scope and machine_id from env vars
         import os
         self.scope = os.environ.get("BUBBALOOP_SCOPE", "local")
@@ -277,8 +296,8 @@ def main():
         "-e",
         "--endpoint",
         type=str,
-        default=None,
-        help="Zenoh endpoint to connect to",
+        default="tcp/127.0.0.1:7447",
+        help="Zenoh endpoint to connect to (default: tcp/127.0.0.1:7447)",
     )
     args = parser.parse_args()
 

network-monitor/pixi.toml

@@ -4,6 +4,7 @@ channels = ["conda-forge"]
 platforms = ["linux-64", "linux-aarch64"]
 
 [tasks]
+build = "python build_proto.py"
 build-proto = "python build_proto.py"
 run = "python main.py -c config.yaml"
 dev = "python main.py -c config.yaml"

openmeteo/Cargo.toml

@@ -22,6 +22,7 @@ serde_json = "1.0"
 serde_yaml = "0.9"
 thiserror = "2.0"
 hostname = "0.4"
+regex-lite = "0.1"
 bubbaloop-schemas = { git = "https://github.com/kornia/bubbaloop.git", branch = "main", features = ["ros-z"] }
 
 [[bin]]

openmeteo/src/api.rs

@@ -120,7 +120,10 @@ pub struct OpenMeteoClient {
 impl OpenMeteoClient {
     pub fn new() -> Self {
         Self {
-            client: reqwest::Client::new(),
+            client: reqwest::Client::builder()
+                .timeout(std::time::Duration::from_secs(30))
+                .build()
+                .expect("Failed to build HTTP client"),
         }
     }
 

openmeteo/src/bin/openmeteo_node.rs

@@ -13,8 +13,8 @@ struct Args {
 
     /// zenoh router endpoint to connect to
     /// Default: tcp/127.0.0.1:7447 (local zenohd router)
-    #[argh(option, short = 'z', default = "String::from(\"tcp/127.0.0.1:7447\")")]
-    zenoh_endpoint: String,
+    #[argh(option, short = 'e', default = "String::from(\"tcp/127.0.0.1:7447\")")]
+    endpoint: String,
 }
 
 #[tokio::main]
@@ -82,7 +82,7 @@ async fn main() -> ZResult<()> {
     log::info!("Scope: {}, Machine ID: {}", scope, machine_id);
 
     // Initialize ROS-Z context
-    let endpoint = std::env::var("ZENOH_ENDPOINT").unwrap_or(args.zenoh_endpoint);
+    let endpoint = std::env::var("ZENOH_ENDPOINT").unwrap_or(args.endpoint);
     log::info!("Connecting to Zenoh at: {}", endpoint);
     let ctx = Arc::new(
         ZContextBuilder::default()

openmeteo/src/config.rs

@@ -91,6 +91,43 @@ pub struct Config {
     pub fetch: FetchConfig,
 }
 
+impl FetchConfig {
+    /// Validate fetch config bounds
+    pub fn validate(&self) -> Result<(), ConfigError> {
+        if self.current_interval_secs == 0 || self.current_interval_secs > 86400 {
+            return Err(ConfigError::ValidationError(format!(
+                "current_interval_secs {} out of range (1-86400)",
+                self.current_interval_secs
+            )));
+        }
+        if self.hourly_interval_secs == 0 || self.hourly_interval_secs > 86400 {
+            return Err(ConfigError::ValidationError(format!(
+                "hourly_interval_secs {} out of range (1-86400)",
+                self.hourly_interval_secs
+            )));
+        }
+        if self.daily_interval_secs == 0 || self.daily_interval_secs > 86400 {
+            return Err(ConfigError::ValidationError(format!(
+                "daily_interval_secs {} out of range (1-86400)",
+                self.daily_interval_secs
+            )));
+        }
+        if self.hourly_forecast_hours == 0 || self.hourly_forecast_hours > 384 {
+            return Err(ConfigError::ValidationError(format!(
+                "hourly_forecast_hours {} out of range (1-384)",
+                self.hourly_forecast_hours
+            )));
+        }
+        if self.daily_forecast_days == 0 || self.daily_forecast_days > 16 {
+            return Err(ConfigError::ValidationError(format!(
+                "daily_forecast_days {} out of range (1-16)",
+                self.daily_forecast_days
+            )));
+        }
+        Ok(())
+    }
+}
+
 impl Config {
     /// Load configuration from a YAML file
     pub fn from_file(path: impl AsRef<Path>) -> Result<Self, ConfigError> {
@@ -101,7 +138,10 @@ impl Config {
 
     /// Parse configuration from a YAML string
     pub fn parse(yaml: &str) -> Result<Self, ConfigError> {
-        serde_yaml::from_str(yaml).map_err(|e| ConfigError::ParseError(e.to_string()))
+        let config: Config =
+            serde_yaml::from_str(yaml).map_err(|e| ConfigError::ParseError(e.to_string()))?;
+        config.fetch.validate()?;
+        Ok(config)
     }
 }
 
@@ -112,4 +152,6 @@ pub enum ConfigError {
     IoError(String),
     #[error("Parse error: {0}")]
     ParseError(String),
+    #[error("Validation error: {0}")]
+    ValidationError(String),
 }