rust/ffi: provide safe flow callback type

Author: jasonish

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

@@ -100,23 +100,26 @@ In Rust, use the ``suricata_ffi::flow`` module:
 - ``flow::register_finish_callback``
 
 The Rust wrappers register closures or function items and return
-``Result<(), &'static str>``.
+``Result<(), &'static str>``. The flow argument is wrapped in
+``flow::Flow`` so safe APIs can be added for common flow operations without
+exposing the entire raw C ``Flow`` layout.
 
 .. code-block:: rust
 
-   use suricata_ffi::flow::{self, Flow, Packet, ThreadVars};
+   use suricata_ffi::flow::{self, Flow};
    use suricata_ffi::SCLogNotice;
+   use suricata_sys::sys::{Packet, ThreadVars};
 
-   fn flow_init(_tv: *mut ThreadVars, f: *mut Flow, _p: *const Packet) {
-       SCLogNotice!("flow initialized: {:p}", f);
+   fn flow_init(_tv: *mut ThreadVars, f: 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: 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: Flow<'_>) {
+       SCLogNotice!("flow finished: {:p}", f.as_ptr());
    }
 
    fn register_flow_callbacks() -> Result<(), &'static str> {
@@ -126,6 +129,6 @@ The Rust wrappers register closures or function items and return
        Ok(())
    }
 
-The raw pointers passed into callbacks are only valid for the duration
-of the callback invocation and must not be stored. Rust callbacks must
-not panic.
+The flow wrapper and raw packet/thread pointers passed into callbacks are only
+valid for the duration of the callback invocation and must not be stored. Rust
+callbacks must not panic.

examples/plugins/rust/src/mod.rs

@@ -64,20 +64,20 @@ fn log_eve_wrapped(
     Ok(())
 }
 
-fn log_flow_init(_tv: *mut ThreadVars, _f: *mut Flow, _p: *const Packet) {
-    SCLogNotice!("rust example flow init callback: flow={:p}", _f);
+fn log_flow_init(_tv: *mut ThreadVars, f: flow::Flow<'_>, _p: *const Packet) {
+    SCLogNotice!("rust example flow init callback: flow={:p}", f.as_ptr());
 }
 
-fn log_flow_update(_tv: *mut ThreadVars, _f: *mut Flow, _p: *mut Packet) {
+fn log_flow_update(_tv: *mut ThreadVars, f: flow::Flow<'_>, p: *mut Packet) {
     SCLogNotice!(
         "rust example flow update callback: flow={:p}, packet={:p}",
-        _f,
-        _p
+        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: flow::Flow<'_>) {
+    SCLogNotice!("rust example flow finish callback: flow={:p}", f.as_ptr());
 }
 
 #[no_mangle]

rust/ffi/src/flow.rs

@@ -15,29 +15,64 @@
  * 02110-1301, USA.
  */
 
+use std::marker::PhantomData;
 use std::os::raw::c_void;
+use std::ptr::NonNull;
 
-use suricata_sys::sys::{Flow, Packet, ThreadVars};
+use suricata_sys::sys::{Flow as CFlow, Packet, ThreadVars};
 use suricata_sys::sys::{
     SCFlowRegisterFinishCallback, SCFlowRegisterInitCallback, SCFlowRegisterUpdateCallback,
 };
 
+/// Access to a Suricata flow for the lifetime of a callback.
+///
+/// This type is intended to be the place where safe Rust accessors for C
+/// `Flow` fields are added. It deliberately does not implement `Deref` or
+/// `DerefMut` to the bindgen type, so field access can be exposed as reviewed
+/// methods instead of exposing the whole raw C layout as safe Rust API.
+pub struct Flow<'a> {
+    ptr: NonNull<CFlow>,
+    _marker: PhantomData<&'a mut CFlow>,
+}
+
+impl<'a> Flow<'a> {
+    /// Create a flow wrapper from a raw pointer.
+    ///
+    /// # Safety
+    ///
+    /// `ptr` must point to a live `Flow` that the caller may access for the
+    /// duration of `'a`. The returned wrapper must not be stored beyond that
+    /// validity period.
+    pub unsafe fn from_ptr(ptr: *mut CFlow) -> Option<Self> {
+        NonNull::new(ptr).map(|ptr| Self {
+            ptr,
+            _marker: PhantomData,
+        })
+    }
+
+    /// Return the wrapped raw C pointer.
+    pub fn as_ptr(&self) -> *const CFlow {
+        self.ptr.as_ptr()
+    }
+
+    /// Return the wrapped raw C pointer as mutable.
+    pub fn as_mut_ptr(&mut self) -> *mut CFlow {
+        self.ptr.as_ptr()
+    }
+}
+
 /// Register a flow initialization callback.
 ///
 /// The callback is invoked whenever Suricata initializes a flow. It receives:
 /// - `tv`: the `ThreadVars` for the thread creating the flow
-/// - `f`: the newly initialized `Flow`
+/// - `flow`: access to the newly initialized `Flow`
 /// - `p`: the packet related to creating the flow
 ///
-/// # Safety
-///
-/// The callback receives raw pointers from Suricata. These pointers are only
-/// valid for the duration of the callback invocation and must not be stored.
-///
-/// The callback must not panic.
+/// The flow wrapper is only valid for the duration of the callback invocation
+/// and must not be stored. The callback must not panic.
 pub fn register_init_callback<F>(callback: F) -> Result<(), &'static str>
 where
-    F: Fn(*mut ThreadVars, *mut Flow, *const Packet) + Send + Sync + 'static,
+    F: for<'a> Fn(*mut ThreadVars, Flow<'a>, *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) } {
@@ -55,18 +90,14 @@ where
 /// The callback is invoked whenever Suricata updates a flow with a packet. It
 /// receives:
 /// - `tv`: the `ThreadVars` for the thread updating the flow
-/// - `f`: the flow being updated
+/// - `flow`: access to the flow being updated
 /// - `p`: the packet responsible for the flow update
 ///
-/// # Safety
-///
-/// The callback receives raw pointers from Suricata. These pointers are only
-/// valid for the duration of the callback invocation and must not be stored.
-///
-/// The callback must not panic.
+/// The flow wrapper is only valid for the duration of the callback invocation
+/// and must not be stored. The callback must not panic.
 pub fn register_update_callback<F>(callback: F) -> Result<(), &'static str>
 where
-    F: Fn(*mut ThreadVars, *mut Flow, *mut Packet) + Send + Sync + 'static,
+    F: for<'a> Fn(*mut ThreadVars, Flow<'a>, *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) } {
@@ -83,17 +114,13 @@ where
 ///
 /// The callback is invoked when Suricata is finished with a flow. It receives:
 /// - `tv`: the `ThreadVars` for the thread finishing the flow
-/// - `f`: the flow being finished
-///
-/// # Safety
-///
-/// The callback receives raw pointers from Suricata. These pointers are only
-/// valid for the duration of the callback invocation and must not be stored.
+/// - `flow`: access to the flow being finished
 ///
-/// The callback must not panic.
+/// The flow wrapper is only valid for the duration of the callback invocation
+/// and must not be stored. The callback must not panic.
 pub fn register_finish_callback<F>(callback: F) -> Result<(), &'static str>
 where
-    F: Fn(*mut ThreadVars, *mut Flow) + Send + Sync + 'static,
+    F: for<'a> Fn(*mut ThreadVars, Flow<'a>) + Send + Sync + 'static,
 {
     let user = Box::into_raw(Box::new(callback)) as *mut c_void;
     if unsafe { SCFlowRegisterFinishCallback(Some(finish_callback_wrapper::<F>), user) } {
@@ -107,28 +134,37 @@ where
 }
 
 unsafe extern "C" fn init_callback_wrapper<F>(
-    tv: *mut ThreadVars, f: *mut Flow, p: *const Packet, user: *mut c_void,
+    tv: *mut ThreadVars, f: *mut CFlow, p: *const Packet, user: *mut c_void,
 ) where
-    F: Fn(*mut ThreadVars, *mut Flow, *const Packet) + Send + Sync + 'static,
+    F: for<'a> Fn(*mut ThreadVars, Flow<'a>, *const Packet) + Send + Sync + 'static,
 {
-    let callback = &*(user as *const F);
-    callback(tv, f, p);
+    let Some(flow) = (unsafe { Flow::from_ptr(f) }) else {
+        return;
+    };
+    let callback = unsafe { &*(user as *const F) };
+    callback(tv, flow, p);
 }
 
 unsafe extern "C" fn update_callback_wrapper<F>(
-    tv: *mut ThreadVars, f: *mut Flow, p: *mut Packet, user: *mut c_void,
+    tv: *mut ThreadVars, f: *mut CFlow, p: *mut Packet, user: *mut c_void,
 ) where
-    F: Fn(*mut ThreadVars, *mut Flow, *mut Packet) + Send + Sync + 'static,
+    F: for<'a> Fn(*mut ThreadVars, Flow<'a>, *mut Packet) + Send + Sync + 'static,
 {
-    let callback = &*(user as *const F);
-    callback(tv, f, p);
+    let Some(flow) = (unsafe { Flow::from_ptr(f) }) else {
+        return;
+    };
+    let callback = unsafe { &*(user as *const F) };
+    callback(tv, flow, p);
 }
 
 unsafe extern "C" fn finish_callback_wrapper<F>(
-    tv: *mut ThreadVars, f: *mut Flow, user: *mut c_void,
+    tv: *mut ThreadVars, f: *mut CFlow, user: *mut c_void,
 ) where
-    F: Fn(*mut ThreadVars, *mut Flow) + Send + Sync + 'static,
+    F: for<'a> Fn(*mut ThreadVars, Flow<'a>) + Send + Sync + 'static,
 {
-    let callback = &*(user as *const F);
-    callback(tv, f);
+    let Some(flow) = (unsafe { Flow::from_ptr(f) }) else {
+        return;
+    };
+    let callback = unsafe { &*(user as *const F) };
+    callback(tv, flow);
 }