rust/ffi: use Flow wrapper in flow and eve callbacks

Pass the safe Flow wrapper to the flow init/update/finish callbacks and
the EVE callback instead of a raw sys::Flow pointer.

Ticket: #8599

doc/userguide/devguide/extending/flow-lifecycle-callbacks.rst

@@ -104,20 +104,21 @@ The Rust wrappers register closures or function items and return
 
 .. code-block:: rust
 
-   use suricata_ffi::flow::{self, Flow, Packet};
+   use suricata_ffi::flow::{self, Flow};
    use suricata_ffi::thread::ThreadVars;
+   use suricata_sys::sys::Packet;
    use suricata_ffi::SCLogNotice;
 
-   fn flow_init(_tv: &mut ThreadVars, f: *mut Flow, _p: *const Packet) {
-       SCLogNotice!("flow initialized: {:p}", f);
+   fn flow_init(_tv: &mut ThreadVars, f: &mut Flow, _p: *const Packet) {
+       SCLogNotice!("flow initialized: {:p}", f.as_ptr());
    }
 
-   fn flow_update(_tv: &mut ThreadVars, f: *mut Flow, p: *mut Packet) {
-       SCLogNotice!("flow updated: {:p} packet: {:p}", f, p);
+   fn flow_update(_tv: &mut ThreadVars, f: &mut Flow, p: *mut Packet) {
+       SCLogNotice!("flow updated: {:p} packet: {:p}", f.as_ptr(), p);
    }
 
-   fn flow_finish(_tv: &mut ThreadVars, f: *mut Flow) {
-       SCLogNotice!("flow finished: {:p}", f);
+   fn flow_finish(_tv: &mut ThreadVars, f: &mut Flow) {
+       SCLogNotice!("flow finished: {:p}", f.as_ptr());
    }
 
    fn register_flow_callbacks() -> Result<(), &'static str> {

examples/plugins/rust/src/mod.rs

@@ -1,11 +1,11 @@
 use std::ptr::null_mut;
 
 use suricata_ffi::eve::{self, SCJsonBuilder};
-use suricata_ffi::flow;
+use suricata_ffi::flow::{self, Flow};
 use suricata_ffi::jsonbuilder::JsonBuilder;
 use suricata_ffi::thread::{self, ThreadStorage, ThreadVars};
 use suricata_ffi::{SCLogError, SCLogNotice, SCLogWarning};
-use suricata_sys::sys::{self, Flow, Packet, SCEveRegisterCallback, SCPlugin};
+use suricata_sys::sys::{self, Packet, SCEveRegisterCallback, SCPlugin};
 
 /// Per-thread state stored in Suricata thread storage.
 #[derive(Default)]
@@ -59,7 +59,7 @@ fn register_thread_callbacks(storage: ThreadStorage<ThreadState>) -> Result<(),
 unsafe extern "C" fn log_eve_raw(
     _tv: *mut sys::ThreadVars,
     _p: *const Packet,
-    _f: *mut Flow,
+    _f: *mut sys::Flow,
     jb: *mut SCJsonBuilder,
     _user: *mut std::os::raw::c_void,
 ) {
@@ -72,12 +72,12 @@ unsafe extern "C" fn log_eve_raw(
 fn log_eve_wrapped(
     _tv: &mut ThreadVars,
     _p: *const Packet,
-    f: *mut Flow,
+    f: Option<&mut Flow>,
     jb: &mut JsonBuilder,
 ) -> Result<(), suricata_ffi::jsonbuilder::Error> {
     jb.open_object("rust_wrapped")?;
     jb.set_string("example", "eve-callback")?;
-    jb.set_string("has_flow", if f.is_null() { "false" } else { "true" })?;
+    jb.set_string("has_flow", if f.is_some() { "true" } else { "false" })?;
     jb.close()?;
     Ok(())
 }
@@ -96,7 +96,7 @@ fn on_thread_init(storage: ThreadStorage<ThreadState>, tv: &mut ThreadVars) {
 fn log_flow_init(
     storage: ThreadStorage<ThreadState>,
     tv: &mut ThreadVars,
-    f: *mut Flow,
+    f: &mut Flow,
     _p: *const Packet,
 ) {
     // Count flows seen by this thread using the per-thread storage.
@@ -112,21 +112,21 @@ fn log_flow_init(
     };
     SCLogNotice!(
         "rust example flow init callback: flow={:p}, thread_flows={}",
-        f,
+        f.as_ptr(),
         flows
     );
 }
 
-fn log_flow_update(_tv: &mut ThreadVars, _f: *mut Flow, _p: *mut Packet) {
+fn log_flow_update(_tv: &mut ThreadVars, f: &mut Flow, _p: *mut Packet) {
     SCLogNotice!(
         "rust example flow update callback: flow={:p}, packet={:p}",
-        _f,
+        f.as_ptr(),
         _p
     );
 }
 
-fn log_flow_finish(_tv: &mut ThreadVars, _f: *mut Flow) {
-    SCLogNotice!("rust example flow finish callback: flow={:p}", _f);
+fn log_flow_finish(_tv: &mut ThreadVars, f: &mut Flow) {
+    SCLogNotice!("rust example flow finish callback: flow={:p}", f.as_ptr());
 }
 
 #[no_mangle]

rust/ffi/src/eve.rs

@@ -23,8 +23,9 @@ use suricata_sys::sys::{
     SCEveFileTypeThreadDeinitFunc, SCEveFileTypeThreadInitFunc, SCEveFileTypeWriteFunc,
     SCEveRegisterCallback, SCRegisterEveFileType,
 };
-pub use suricata_sys::sys::{Flow, Packet, SCEveUserCallbackFn, SCJsonBuilder};
+pub use suricata_sys::sys::{Packet, SCEveUserCallbackFn, SCJsonBuilder};
 
+use crate::flow::Flow;
 use crate::thread::ThreadVars;
 
 pub struct EveFileType {
@@ -82,7 +83,7 @@ impl EveFileType {
 /// The callback receives:
 /// - `tv`: the `ThreadVars` for the thread performing the logging
 /// - `p`: the `Packet`, if available
-/// - `f`: the `Flow`, if available
+/// - `f`: the `Flow`, if available (`None` when no flow is associated)
 /// - `jb`: the JSON builder for the current EVE record
 ///
 /// This API is intended for plugin and library users.
@@ -115,7 +116,7 @@ where
     F: Fn(
             &mut ThreadVars,
             *const Packet,
-            *mut Flow,
+            Option<&mut Flow>,
             &mut crate::jsonbuilder::JsonBuilder,
         ) -> Result<(), crate::jsonbuilder::Error>
         + Send
@@ -136,13 +137,13 @@ where
 /// Internal wrapper used to adapt the C EVE callback to a Rust
 /// closure callback.
 unsafe extern "C" fn callback_wrapper<F>(
-    tv: *mut sys::ThreadVars, p: *const Packet, f: *mut Flow, jb: *mut SCJsonBuilder,
+    tv: *mut sys::ThreadVars, p: *const Packet, f: *mut sys::Flow, jb: *mut SCJsonBuilder,
     user: *mut c_void,
 ) where
     F: Fn(
             &mut ThreadVars,
             *const Packet,
-            *mut Flow,
+            Option<&mut Flow>,
             &mut crate::jsonbuilder::JsonBuilder,
         ) -> Result<(), crate::jsonbuilder::Error>
         + Send
@@ -151,9 +152,14 @@ unsafe extern "C" fn callback_wrapper<F>(
 {
     let callback = &*(user as *const F);
     let mut tv = ThreadVars::from_ptr(tv);
+    let mut flow = if f.is_null() {
+        None
+    } else {
+        Some(Flow::from_ptr(f))
+    };
     let mut jb = crate::jsonbuilder::JsonBuilder::from_raw(jb);
     let mark = jb.get_mark();
-    if callback(&mut tv, p, f, &mut jb).is_err() {
+    if callback(&mut tv, p, flow.as_mut(), &mut jb).is_err() {
         let _ = jb.restore_mark(&mark);
     }
 }

rust/ffi/src/flow.rs

@@ -69,7 +69,7 @@ impl<'a> Flow<'a> {
 /// The callback must not panic.
 pub fn register_init_callback<F>(callback: F) -> Result<(), &'static str>
 where
-    F: Fn(&mut ThreadVars, *mut sys::Flow, *const Packet) + Send + Sync + 'static,
+    F: Fn(&mut ThreadVars, &mut Flow, *const Packet) + Send + Sync + 'static,
 {
     let user = Box::into_raw(Box::new(callback)) as *mut c_void;
     if unsafe { SCFlowRegisterInitCallback(Some(init_callback_wrapper::<F>), user) } {
@@ -98,7 +98,7 @@ where
 /// The callback must not panic.
 pub fn register_update_callback<F>(callback: F) -> Result<(), &'static str>
 where
-    F: Fn(&mut ThreadVars, *mut sys::Flow, *mut Packet) + Send + Sync + 'static,
+    F: Fn(&mut ThreadVars, &mut Flow, *mut Packet) + Send + Sync + 'static,
 {
     let user = Box::into_raw(Box::new(callback)) as *mut c_void;
     if unsafe { SCFlowRegisterUpdateCallback(Some(update_callback_wrapper::<F>), user) } {
@@ -125,7 +125,7 @@ where
 /// The callback must not panic.
 pub fn register_finish_callback<F>(callback: F) -> Result<(), &'static str>
 where
-    F: Fn(&mut ThreadVars, *mut sys::Flow) + Send + Sync + 'static,
+    F: Fn(&mut ThreadVars, &mut Flow) + Send + Sync + 'static,
 {
     let user = Box::into_raw(Box::new(callback)) as *mut c_void;
     if unsafe { SCFlowRegisterFinishCallback(Some(finish_callback_wrapper::<F>), user) } {
@@ -141,29 +141,32 @@ where
 unsafe extern "C" fn init_callback_wrapper<F>(
     tv: *mut sys::ThreadVars, f: *mut sys::Flow, p: *const Packet, user: *mut c_void,
 ) where
-    F: Fn(&mut ThreadVars, *mut sys::Flow, *const Packet) + Send + Sync + 'static,
+    F: Fn(&mut ThreadVars, &mut Flow, *const Packet) + Send + Sync + 'static,
 {
     let callback = &*(user as *const F);
     let mut tv = ThreadVars::from_ptr(tv);
-    callback(&mut tv, f, p);
+    let mut f = Flow::from_ptr(f);
+    callback(&mut tv, &mut f, p);
 }
 
 unsafe extern "C" fn update_callback_wrapper<F>(
     tv: *mut sys::ThreadVars, f: *mut sys::Flow, p: *mut Packet, user: *mut c_void,
 ) where
-    F: Fn(&mut ThreadVars, *mut sys::Flow, *mut Packet) + Send + Sync + 'static,
+    F: Fn(&mut ThreadVars, &mut Flow, *mut Packet) + Send + Sync + 'static,
 {
     let callback = &*(user as *const F);
     let mut tv = ThreadVars::from_ptr(tv);
-    callback(&mut tv, f, p);
+    let mut f = Flow::from_ptr(f);
+    callback(&mut tv, &mut f, p);
 }
 
 unsafe extern "C" fn finish_callback_wrapper<F>(
     tv: *mut sys::ThreadVars, f: *mut sys::Flow, user: *mut c_void,
 ) where
-    F: Fn(&mut ThreadVars, *mut sys::Flow) + Send + Sync + 'static,
+    F: Fn(&mut ThreadVars, &mut Flow) + Send + Sync + 'static,
 {
     let callback = &*(user as *const F);
     let mut tv = ThreadVars::from_ptr(tv);
-    callback(&mut tv, f);
+    let mut f = Flow::from_ptr(f);
+    callback(&mut tv, &mut f);
 }