netget/server/ipfix/
mod.rs

1//! Receive/parse ownership stays independent of parked common handlers.
2pub mod actions;
3pub mod codec;
4use crate::protocol::{Event, SpawnContext};
5use crate::server::connection::ConnectionId;
6use crate::state::{
7    server::{ConnectionState, ConnectionStatus, ProtocolConnectionInfo},
8    AccessLogOwner,
9};
10use crate::{console_error, console_info};
11use anyhow::{ensure, Result};
12use serde_json::json;
13use std::{net::SocketAddr, time::Duration};
14pub const MAX_QUEUED_EVENTS: usize = 32;
15pub struct IpfixServer;
16pub fn duration(value: u64) -> Result<Duration> {
17    ensure!(
18        (1..=86400).contains(&value),
19        "IPFIX duration must be1..86400seconds"
20    );
21    Ok(Duration::from_secs(value))
22}
23impl IpfixServer {
24    pub async fn spawn(ctx: SpawnContext) -> Result<SocketAddr> {
25        let params = ctx.startup_params.as_ref();
26        let ttl = duration(
27            params
28                .map(|p| p.get_optional_u64("template_ttl_seconds"))
29                .transpose()?
30                .flatten()
31                .unwrap_or(codec::DEFAULT_TEMPLATE_TTL),
32        )?;
33        let idle = duration(
34            params
35                .map(|p| p.get_optional_u64("session_idle_seconds"))
36                .transpose()?
37                .flatten()
38                .unwrap_or(codec::DEFAULT_SESSION_IDLE),
39        )?;
40        let llm = params
41            .map(|p| p.get_optional_bool("llm_fallback"))
42            .transpose()?
43            .flatten()
44            .unwrap_or(codec::DEFAULT_LLM_FALLBACK);
45        let socket = tokio::net::UdpSocket::bind(ctx.legacy_listen_addr()).await?;
46        let local = socket.local_addr()?;
47        console_info!(
48            ctx.status_tx,
49            "IPFIX UDP listening on {} (llm_fallback={})",
50            local,
51            llm
52        );
53        let (tx, mut rx) =
54            tokio::sync::mpsc::channel::<(SocketAddr, usize, codec::Message)>(MAX_QUEUED_EVENTS);
55        let dispatcher = ctx.clone();
56        let owner = ctx.state.clone();
57        let id = ctx.server_id;
58        owner
59            .spawn_server_task(id, async move {
60                while let Some((peer, n, message)) = rx.recv().await {
61                    let cid = ConnectionId::new(dispatcher.state.get_next_unified_id().await);
62                    let now = crate::utils::clock::Instant::now();
63                    dispatcher
64                        .state
65                        .add_connection_to_server(
66                            id,
67                            ConnectionState {
68                                id: cid,
69                                remote_addr: peer,
70                                local_addr: local,
71                                bytes_sent: 0,
72                                bytes_received: n as u64,
73                                packets_sent: 0,
74                                packets_received: 1,
75                                last_activity: now,
76                                status: ConnectionStatus::Active,
77                                status_changed_at: now,
78                                protocol_info: ProtocolConnectionInfo::empty(),
79                            },
80                        )
81                        .await;
82                    let event = Event::new(
83                        &actions::IPFIX_MESSAGE_EVENT,
84                        json!({"message":message, "source_addr":peer.to_string()}),
85                    );
86                    let configured = dispatcher
87                        .state
88                        .get_event_handler_config(id)
89                        .await
90                        .is_some_and(|c| c.find_handler("ipfix_message").is_some());
91                    if llm || configured {
92                        match crate::llm::action_helper::call_llm(
93                            &dispatcher.llm_client,
94                            &dispatcher.state,
95                            id,
96                            Some(cid),
97                            &event,
98                            &actions::IpfixProtocol::new(),
99                        )
100                        .await
101                        {
102                            Ok(result) if result.failures.is_empty() => {
103                                let decision = if result
104                                    .protocol_result_actions
105                                    .iter()
106                                    .any(|action| action == "collect_ipfix_records")
107                                {
108                                    "handler_collect"
109                                } else {
110                                    "handler_silent"
111                                };
112                                for m in result.messages {
113                                    console_info!(dispatcher.status_tx, "{}", m);
114                                }
115                                console_info!(
116                                    dispatcher.status_tx,
117                                    "IPFIX source={} decision={} udp_silent=true",
118                                    peer,
119                                    decision
120                                );
121                                dispatcher
122                                    .state
123                                    .record_access_log(
124                                        AccessLogOwner::Server(id.as_u32()),
125                                        "IPFIX",
126                                        Some(cid.as_u32()),
127                                        "ipfix_handler_decision",
128                                        event.data,
129                                        vec![json!({"decision":decision,"udp_silent":true})],
130                                    )
131                                    .await;
132                            }
133                            result => {
134                                let (decision, reason) = match result {
135                                    Ok(r) => (
136                                        "fail_closed_action_error",
137                                        format!("{} failed actions", r.failures.len()),
138                                    ),
139                                    Err(e) => ("fail_closed_dispatch_error", e.to_string()),
140                                };
141                                console_error!(
142                                    dispatcher.status_tx,
143                                    "IPFIX source={} decision={} udp_silent=true error={}",
144                                    peer,
145                                    decision,
146                                    reason
147                                );
148                                dispatcher
149                                    .state
150                                    .record_access_log(
151                                        AccessLogOwner::Server(id.as_u32()),
152                                        "IPFIX",
153                                        Some(cid.as_u32()),
154                                        "ipfix_handler_failed",
155                                        event.data,
156                                        vec![json!({"decision":"fail_closed_handler_error", "error":reason,"terminal_decision":decision,"udp_silent":true})],
157                                    )
158                                    .await;
159                            }
160                        }
161                    } else {
162                        console_info!(
163                            dispatcher.status_tx,
164                            "IPFIX source={} decision=default_collect udp_silent=true",
165                            peer
166                        );
167                        dispatcher
168                            .state
169                            .record_access_log(
170                                AccessLogOwner::Server(id.as_u32()),
171                                "IPFIX",
172                                Some(cid.as_u32()),
173                                "ipfix_message",
174                                event.data,
175                                vec![json!({"type":"collect_ipfix_records","decision":"default_collect","udp_silent":true})],
176                            )
177                            .await;
178                    }
179                    dispatcher
180                        .state
181                        .remove_connection_from_server(id, cid)
182                        .await;
183                    let _ = dispatcher.status_tx.send("__UPDATE_UI__".into());
184                }
185            })
186            .await;
187        owner
188            .spawn_server_task(id, async move {
189                let mut cache = codec::TemplateCache::new(ttl, idle);
190                let mut buffer = vec![0; codec::MAX_MESSAGE_BYTES + 1];
191                let mut expiry = tokio::time::interval(Duration::from_secs(1));
192                expiry.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
193                loop {
194                    let (n, peer) = tokio::select! {
195                        _ = expiry.tick() => {
196                            cache.expire(tokio::time::Instant::now());
197                            continue;
198                        },
199                        r = socket.recv_from(&mut buffer) => match r {
200                            Ok(v) => v,
201                            Err(e) => {
202                                console_error!(ctx.status_tx, "IPFIX decision=fail_closed_receive_error udp_silent=true error={}", e);
203                                break;
204                            }
205                        }
206                    };
207                    match cache.ingest(peer, &buffer[..n], tokio::time::Instant::now()) {
208                        Ok(message) => {
209                            if let Err(error) = tx.try_send((peer, n, message)) {
210                                let closed = matches!(
211                                    error,
212                                    tokio::sync::mpsc::error::TrySendError::Closed(_)
213                                );
214                                let decision = if closed {
215                                    "fail_closed_dispatch_closed"
216                                } else {
217                                    "fail_closed_event_capacity"
218                                };
219                                console_error!(
220                                    ctx.status_tx,
221                                    "IPFIX source={} decision={} udp_silent=true event_cap={}",
222                                    peer,
223                                    decision,
224                                    MAX_QUEUED_EVENTS
225                                );
226                                ctx.state
227                                    .record_access_log(
228                                        AccessLogOwner::Server(id.as_u32()),
229                                        "IPFIX",
230                                        None,
231                                        "ipfix_event_capacity",
232                                        json!({"source_addr":peer.to_string()}),
233                                        vec![json!({"decision":"fail_closed_event_capacity", "event_cap":MAX_QUEUED_EVENTS,"terminal_decision":decision,"udp_silent":true})],
234                                    )
235                                    .await;
236                                if closed {
237                                    break;
238                                }
239                            }
240                        }
241                        Err(error) => {
242                            console_error!(
243                                ctx.status_tx,
244                                "IPFIX source={} decision=fail_closed_invalid_datagram udp_silent=true received_bytes={} error={}",
245                                peer,
246                                n,
247                                error
248                            );
249                            ctx.state
250                                .record_access_log(
251                                    AccessLogOwner::Server(id.as_u32()),
252                                    "IPFIX",
253                                    None,
254                                    "ipfix_invalid_datagram",
255                                    json!({"source_addr":peer.to_string(), "received_bytes":n}),
256                                    vec![json!({"decision":"fail_closed_invalid_datagram", "error":error.to_string()})],
257                                )
258                                .await;
259                        }
260                    }
261                }
262            })
263            .await;
264        Ok(local)
265    }
266}
netget/server/ipfix/mod.rs

netget/server/ipfix/
mod.rs
