Skip to content

Commit bdbd13f

Browse files
committed
rust/ffi: add safe flow storage wrapper
Add a typed FlowStorage<T> wrapper around the flow storage bindings. Update example and docs. Ticket: OISF#8447
1 parent 91087e1 commit bdbd13f

3 files changed

Lines changed: 221 additions & 18 deletions

File tree

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

Lines changed: 42 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -131,3 +131,45 @@ The Rust wrappers register closures or function items and return
131131
The raw pointers passed into callbacks are only valid for the duration
132132
of the callback invocation and must not be stored. Rust callbacks must
133133
not panic.
134+
135+
Flow Storage
136+
============
137+
138+
``flow::FlowStorage<T>`` provides typed, per-flow storage backed by
139+
Suricata's flow storage API. Each registered slot holds an independent value
140+
of type ``T`` for every flow.
141+
142+
Register a slot once during initialization with
143+
``FlowStorage::<T>::register``. Registration must happen before Suricata
144+
finalizes its storage registration, which is the case during plugin
145+
initialization.
146+
147+
.. code-block:: rust
148+
149+
use suricata_ffi::flow::{self, Flow, FlowStorage};
150+
use suricata_ffi::thread::ThreadVars;
151+
use suricata_sys::sys::Packet;
152+
153+
#[derive(Default)]
154+
struct FlowState {
155+
packets: u64,
156+
}
157+
158+
fn register(storage: FlowStorage<FlowState>) -> Result<(), &'static str> {
159+
flow::register_update_callback(move |tv, f, p| on_flow_update(storage, tv, f, p))
160+
}
161+
162+
Values are owned by Suricata's flow storage and are dropped automatically when
163+
the flow's storage is freed.
164+
165+
.. code-block:: rust
166+
167+
fn on_flow_init(storage: FlowStorage<FlowState>, _tv: &mut ThreadVars, f: &mut Flow, _p: *const Packet) {
168+
let _ = storage.get_or_insert_with(f, FlowState::default);
169+
}
170+
171+
fn on_flow_update(storage: FlowStorage<FlowState>, _tv: &mut ThreadVars, f: &mut Flow, _p: *mut Packet) {
172+
if let Some(state) = storage.get_mut(f) {
173+
state.packets += 1;
174+
}
175+
}

examples/plugins/rust/src/mod.rs

Lines changed: 68 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
use std::ptr::null_mut;
22

33
use suricata_ffi::eve::{self, SCJsonBuilder};
4-
use suricata_ffi::flow::{self, Flow};
4+
use suricata_ffi::flow::{self, Flow, FlowStorage};
55
use suricata_ffi::jsonbuilder::JsonBuilder;
66
use suricata_ffi::thread::{self, ThreadStorage, ThreadVars};
77
use suricata_ffi::{SCLogError, SCLogNotice, SCLogWarning};
@@ -13,6 +13,12 @@ struct ThreadState {
1313
flows: u64,
1414
}
1515

16+
/// Per-flow state stored in Suricata flow storage.
17+
#[derive(Default)]
18+
struct FlowState {
19+
packets: u64,
20+
}
21+
1622
unsafe extern "C" fn init() {
1723
suricata_ffi::plugin::init();
1824
SCLogNotice!("Initializing rust example plugin");
@@ -26,29 +32,41 @@ unsafe extern "C" fn init() {
2632
return;
2733
}
2834
};
35+
let flow_storage = match FlowStorage::<FlowState>::register("rust-example-flow") {
36+
Ok(storage) => storage,
37+
Err(err) => {
38+
SCLogError!("Failed to register rust example flow storage: {}", err);
39+
return;
40+
}
41+
};
2942

30-
if let Err(err) = register_eve_callbacks() {
43+
if let Err(err) = register_eve_callbacks(flow_storage) {
3144
SCLogError!("Failed to register rust example EVE callbacks: {}", err);
3245
}
33-
if let Err(err) = register_flow_callbacks(thread_storage) {
46+
if let Err(err) = register_flow_callbacks(thread_storage, flow_storage) {
3447
SCLogError!("Failed to register rust example flow callbacks: {}", err);
3548
}
3649
if let Err(err) = register_thread_callbacks(thread_storage) {
3750
SCLogError!("Failed to register rust example thread callbacks: {}", err);
3851
}
3952
}
4053

41-
fn register_eve_callbacks() -> Result<(), &'static str> {
54+
fn register_eve_callbacks(flow_storage: FlowStorage<FlowState>) -> Result<(), &'static str> {
4255
if !unsafe { SCEveRegisterCallback(Some(log_eve_raw), null_mut()) } {
4356
return Err("Failed to register raw EVE callback");
4457
}
45-
eve::register_callback(log_eve_wrapped)
58+
eve::register_callback(move |tv, p, f, jb| log_eve_wrapped(flow_storage, tv, p, f, jb))
4659
}
4760

48-
fn register_flow_callbacks(storage: ThreadStorage<ThreadState>) -> Result<(), &'static str> {
49-
flow::register_init_callback(move |tv, f, p| log_flow_init(storage, tv, f, p))?;
50-
flow::register_update_callback(log_flow_update)?;
51-
flow::register_finish_callback(log_flow_finish)?;
61+
fn register_flow_callbacks(
62+
thread_storage: ThreadStorage<ThreadState>,
63+
flow_storage: FlowStorage<FlowState>,
64+
) -> Result<(), &'static str> {
65+
flow::register_init_callback(move |tv, f, p| {
66+
log_flow_init(thread_storage, flow_storage, tv, f, p)
67+
})?;
68+
flow::register_update_callback(move |tv, f, p| log_flow_update(flow_storage, tv, f, p))?;
69+
flow::register_finish_callback(move |tv, f| log_flow_finish(flow_storage, tv, f))?;
5270
Ok(())
5371
}
5472

@@ -70,6 +88,7 @@ unsafe extern "C" fn log_eve_raw(
7088
}
7189

7290
fn log_eve_wrapped(
91+
flow_storage: FlowStorage<FlowState>,
7392
_tv: &mut ThreadVars,
7493
_p: *const Packet,
7594
f: Option<&mut Flow>,
@@ -78,6 +97,13 @@ fn log_eve_wrapped(
7897
jb.open_object("rust_wrapped")?;
7998
jb.set_string("example", "eve-callback")?;
8099
jb.set_string("has_flow", if f.is_some() { "true" } else { "false" })?;
100+
101+
// If we have a flow, log something from flow storage.
102+
if let Some(f) = f {
103+
if let Some(state) = flow_storage.get(f) {
104+
jb.set_uint("flow_packets", state.packets)?;
105+
}
106+
}
81107
jb.close()?;
82108
Ok(())
83109
}
@@ -94,13 +120,14 @@ fn on_thread_init(storage: ThreadStorage<ThreadState>, tv: &mut ThreadVars) {
94120
}
95121

96122
fn log_flow_init(
97-
storage: ThreadStorage<ThreadState>,
123+
thread_storage: ThreadStorage<ThreadState>,
124+
flow_storage: FlowStorage<FlowState>,
98125
tv: &mut ThreadVars,
99126
f: &mut Flow,
100127
_p: *const Packet,
101128
) {
102129
// Count flows seen by this thread using the per-thread storage.
103-
let flows = match storage.get_mut(tv) {
130+
let flows = match thread_storage.get_mut(tv) {
104131
Some(state) => {
105132
state.flows += 1;
106133
state.flows
@@ -110,23 +137,48 @@ fn log_flow_init(
110137
0
111138
}
112139
};
140+
// Initialize the per-flow storage for this flow.
141+
if let Err(err) = flow_storage.get_or_insert_with(f, FlowState::default) {
142+
SCLogError!("failed to initialize rust example flow storage: {}", err);
143+
}
113144
SCLogNotice!(
114145
"rust example flow init callback: flow={:p}, thread_flows={}",
115146
f.as_ptr(),
116147
flows
117148
);
118149
}
119150

120-
fn log_flow_update(_tv: &mut ThreadVars, f: &mut Flow, _p: *mut Packet) {
151+
fn log_flow_update(
152+
flow_storage: FlowStorage<FlowState>,
153+
_tv: &mut ThreadVars,
154+
f: &mut Flow,
155+
_p: *mut Packet,
156+
) {
157+
// Count packets seen on this flow using the per-flow storage.
158+
let packets = match flow_storage.get_mut(f) {
159+
Some(state) => {
160+
state.packets += 1;
161+
state.packets
162+
}
163+
None => {
164+
SCLogWarning!("rust example flow storage was not initialized");
165+
0
166+
}
167+
};
121168
SCLogNotice!(
122-
"rust example flow update callback: flow={:p}, packet={:p}",
169+
"rust example flow update callback: flow={:p}, flow_packets={}",
123170
f.as_ptr(),
124-
_p
171+
packets
125172
);
126173
}
127174

128-
fn log_flow_finish(_tv: &mut ThreadVars, f: &mut Flow) {
129-
SCLogNotice!("rust example flow finish callback: flow={:p}", f.as_ptr());
175+
fn log_flow_finish(flow_storage: FlowStorage<FlowState>, _tv: &mut ThreadVars, f: &mut Flow) {
176+
let packets = flow_storage.get(f).map(|state| state.packets).unwrap_or(0);
177+
SCLogNotice!(
178+
"rust example flow finish callback: flow={:p}, flow_packets={}",
179+
f.as_ptr(),
180+
packets
181+
);
130182
}
131183

132184
#[no_mangle]

rust/ffi/src/flow.rs

Lines changed: 111 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,4 @@
1+
12
/* Copyright (C) 2026 Open Information Security Foundation
23
*
34
* You can copy, redistribute or modify this Program under the terms of
@@ -15,12 +16,13 @@
1516
* 02110-1301, USA.
1617
*/
1718

19+
use std::ffi::CString;
1820
use std::marker::PhantomData;
1921
use std::os::raw::c_void;
2022

2123
use suricata_sys::sys::{
22-
self, Packet, SCFlowRegisterFinishCallback, SCFlowRegisterInitCallback,
23-
SCFlowRegisterUpdateCallback,
24+
self, Packet, SCFlowGetStorageById, SCFlowRegisterFinishCallback, SCFlowRegisterInitCallback,
25+
SCFlowRegisterUpdateCallback, SCFlowSetStorageById, SCFlowStorageId, SCFlowStorageRegister,
2426
};
2527

2628
use crate::thread::ThreadVars;
@@ -62,6 +64,113 @@ impl<'a> Flow<'a> {
6264
}
6365
}
6466

67+
/// A typed handle to a per-flow storage slot.
68+
///
69+
/// `FlowStorage<T>` wraps the `SCFlowStorageId` returned when registering flow
70+
/// storage with Suricata. Values are stored as a `Box<T>` owned by Suricata's
71+
/// flow storage and are dropped automatically when the flow's storage is freed.
72+
///
73+
/// The handle only holds the storage id, so it is `Copy` and `Send`/`Sync`
74+
/// regardless of `T`, and can be passed by value into the callbacks that need
75+
/// it.
76+
pub struct FlowStorage<T> {
77+
id: SCFlowStorageId,
78+
_marker: PhantomData<fn() -> T>,
79+
}
80+
81+
// Manual `Copy`/`Clone` impls so the handle is copyable regardless of whether
82+
// `T` is; it only holds the storage id.
83+
impl<T> Clone for FlowStorage<T> {
84+
fn clone(&self) -> Self {
85+
*self
86+
}
87+
}
88+
89+
impl<T> Copy for FlowStorage<T> {}
90+
91+
impl<T: Send + 'static> FlowStorage<T> {
92+
/// Register a new flow storage slot for values of type `T`.
93+
///
94+
/// `name` must be unique among registered flow storage. Registration has to
95+
/// happen during initialization, before Suricata finalizes storage
96+
/// registration (`SCStorageFinalize`).
97+
///
98+
/// Returns an error if `name` contains an interior nul byte or if Suricata
99+
/// rejects the registration.
100+
pub fn register(name: &str) -> Result<Self, &'static str> {
101+
let name = CString::new(name).map_err(|_| "flow storage name contains a nul byte")?;
102+
let id = unsafe { SCFlowStorageRegister(name.as_ptr(), Some(Self::free)) };
103+
if id.id < 0 {
104+
return Err("Failed to register flow storage");
105+
}
106+
107+
// Suricata keeps the storage name pointer in its storage mapping for
108+
// the lifetime of the process, so the CString is intentionally leaked.
109+
std::mem::forget(name);
110+
111+
Ok(Self {
112+
id,
113+
_marker: PhantomData,
114+
})
115+
}
116+
117+
/// Return a reference to the value stored for `f`, if any.
118+
pub fn get<'f>(&self, f: &'f Flow<'_>) -> Option<&'f T> {
119+
let ptr = unsafe { SCFlowGetStorageById(f.as_ptr(), self.id) };
120+
if ptr.is_null() {
121+
None
122+
} else {
123+
Some(unsafe { &*(ptr as *const T) })
124+
}
125+
}
126+
127+
/// Return a mutable reference to the value stored for `f`, if any.
128+
pub fn get_mut<'f>(&self, f: &'f mut Flow<'_>) -> Option<&'f mut T> {
129+
let ptr = unsafe { SCFlowGetStorageById(f.as_ptr(), self.id) };
130+
if ptr.is_null() {
131+
None
132+
} else {
133+
Some(unsafe { &mut *(ptr as *mut T) })
134+
}
135+
}
136+
137+
/// Return a mutable reference to the value stored for `f`, inserting the
138+
/// value produced by `init` if none is present yet.
139+
pub fn get_or_insert_with<'f>(
140+
&self, f: &'f mut Flow<'_>, init: impl FnOnce() -> T,
141+
) -> Result<&'f mut T, &'static str> {
142+
let ptr = unsafe { SCFlowGetStorageById(f.as_ptr(), self.id) };
143+
if !ptr.is_null() {
144+
return Ok(unsafe { &mut *(ptr as *mut T) });
145+
}
146+
147+
// `SCFlowSetStorageById` overwrites the slot without freeing any
148+
// previous value; we only reach here when the slot is empty.
149+
let ptr = Box::into_raw(Box::new(init()));
150+
let rc = unsafe { SCFlowSetStorageById(f.as_mut_ptr(), self.id, ptr.cast()) };
151+
if rc != 0 {
152+
unsafe {
153+
drop(Box::from_raw(ptr));
154+
}
155+
return Err("Failed to set flow storage");
156+
}
157+
158+
Ok(unsafe { &mut *ptr })
159+
}
160+
161+
/// Free callback registered with Suricata flow storage that drops the
162+
/// `Box<T>` backing a stored value.
163+
unsafe extern "C" fn free(ptr: *mut c_void) {
164+
if !ptr.is_null() {
165+
// The drop runs across an FFI boundary, so guard against unwinding
166+
// into C if `T`'s `Drop` panics.
167+
let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
168+
drop(Box::from_raw(ptr as *mut T));
169+
}));
170+
}
171+
}
172+
}
173+
65174
/// Register a flow initialization callback.
66175
///
67176
/// The callback is invoked whenever Suricata initializes a flow. It receives:

0 commit comments

Comments
 (0)