migration: phase 1 — restore local gRPC ingest in syva-core

syva-core/Cargo.toml

@@ -13,7 +13,9 @@ syva-cp-client = { path = "../syva-cp-client" }
 syva-ebpf-common = { path = "../syva-ebpf-common", features = ["userspace"] }
 aya = { workspace = true }
 tokio = { workspace = true }
+tokio-stream = { workspace = true, features = ["net"] }
 tokio-util = { workspace = true }
+tonic = { workspace = true }
 tracing = { workspace = true }
 tracing-subscriber = { workspace = true }
 anyhow = { workspace = true }
@@ -23,3 +25,8 @@ serde_json = { workspace = true }
 clap = { workspace = true }
 libc = "0.2"
 uuid = { version = "1", features = ["v4", "serde"] }
+
+[dev-dependencies]
+hyper-util = { version = "0.1", features = ["tokio"] }
+tempfile = "3"
+tower = { version = "0.4", features = ["util"] }

syva-core/src/container_id.rs

@@ -0,0 +1,29 @@
+/// Validate that a container ID is safe to accept from local adapters.
+pub(crate) fn is_valid_container_id(id: &str) -> bool {
+    !id.is_empty()
+        && id.len() <= 128
+        && id
+            .bytes()
+            .all(|b| b.is_ascii_hexdigit() || b == b'-' || b == b'_')
+}
+
+#[cfg(test)]
+mod tests {
+    use super::is_valid_container_id;
+
+    #[test]
+    fn accepts_containerd_like_ids() {
+        assert!(is_valid_container_id("abc123"));
+        assert!(is_valid_container_id("abc-def_123"));
+        assert!(is_valid_container_id(&"a".repeat(128)));
+    }
+
+    #[test]
+    fn rejects_path_or_shell_shaped_ids() {
+        assert!(!is_valid_container_id(""));
+        assert!(!is_valid_container_id("../../../etc/passwd"));
+        assert!(!is_valid_container_id("abc/def"));
+        assert!(!is_valid_container_id("abc def"));
+        assert!(!is_valid_container_id(&"a".repeat(129)));
+    }
+}

syva-core/src/ingest.rs

@@ -34,6 +34,9 @@ pub(crate) async fn register_zone_local(
         let was_new = registry.zone_id(zone_name).is_none();
         registry.register_zone(zone_name)?;
         let zone_id = registry.revive_draining(zone_name)?;
+        if let Some(policy) = policy.as_ref() {
+            registry.set_zone_type(zone_name, policy.zone_type)?;
+        }
         let zones_loaded = registry.zone_count();
         (zone_id, zones_loaded, was_new)
     };
@@ -71,8 +74,6 @@ pub(crate) async fn register_zone_local(
                 }
             }
         }
-
-        let _ = policy.zone_type;
     }
 
     health.write().await.zones_loaded = zones_loaded;