From 8cf5d4d7703d5f73273e3e76fc4105233400d721 Mon Sep 17 00:00:00 2001 From: Jason Ish Date: Fri, 29 May 2026 16:57:06 -0600 Subject: [PATCH] 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 --- .../extending/flow-lifecycle-callbacks.rst | 15 +++++++------ examples/plugins/rust/src/mod.rs | 22 +++++++++---------- rust/ffi/src/eve.rs | 18 ++++++++++----- rust/ffi/src/flow.rs | 21 ++++++++++-------- 4 files changed, 43 insertions(+), 33 deletions(-) diff --git a/doc/userguide/devguide/extending/flow-lifecycle-callbacks.rst b/doc/userguide/devguide/extending/flow-lifecycle-callbacks.rst index a11f945fc3..139028548c 100644 --- a/doc/userguide/devguide/extending/flow-lifecycle-callbacks.rst +++ b/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> { diff --git a/examples/plugins/rust/src/mod.rs b/examples/plugins/rust/src/mod.rs index 22cccb7c52..40026df44e 100644 --- a/examples/plugins/rust/src/mod.rs +++ b/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) -> 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, tv: &mut ThreadVars) { fn log_flow_init( storage: ThreadStorage, 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] diff --git a/rust/ffi/src/eve.rs b/rust/ffi/src/eve.rs index 8440c9d3fd..6b5514659b 100644 --- a/rust/ffi/src/eve.rs +++ b/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( - 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( { 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); } } diff --git a/rust/ffi/src/flow.rs b/rust/ffi/src/flow.rs index 3f99252b0e..990d1a36a9 100644 --- a/rust/ffi/src/flow.rs +++ b/rust/ffi/src/flow.rs @@ -69,7 +69,7 @@ impl<'a> Flow<'a> { /// The callback must not panic. pub fn register_init_callback(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::), user) } { @@ -98,7 +98,7 @@ where /// The callback must not panic. pub fn register_update_callback(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::), user) } { @@ -125,7 +125,7 @@ where /// The callback must not panic. pub fn register_finish_callback(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::), user) } { @@ -141,29 +141,32 @@ where unsafe extern "C" fn init_callback_wrapper( 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( 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( 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); }