1use crate::llm::actions::{
2 client_trait::{Client, ClientActionResult},
3 protocol_trait::Protocol,
4 ActionDefinition, ParameterDefinition, StartupExamples,
5};
6use crate::protocol::log_template::LogTemplate;
7use crate::protocol::{
8 metadata::{DevelopmentState, ProtocolMetadataV2},
9 ConnectContext, EventType,
10};
11use crate::server::ipfix::{
12 actions::{duration_parameter, parameter},
13 codec::{self, Batch},
14};
15use crate::state::app_state::AppState;
16use anyhow::{bail, Context, Result};
17use serde_json::{json, Value};
18use std::sync::LazyLock;
19#[derive(Default)]
20pub struct IpfixClientProtocol;
21impl IpfixClientProtocol {
22 pub fn new() -> Self {
23 Self
24 }
25}
26pub fn example_batch() -> Value {
27 json!({"observation_domain_id":42,"templates":[{"id":256,"fields":[{"element":"source_ipv4_address"},{"element":"destination_ipv4_address"},{"element":"packet_delta_count"}]}],"data_sets":[{"template_id":256,"records":[[{"kind":"ipv4","value":"192.0.2.1"},{"kind":"ipv4","value":"198.51.100.2"},{"kind":"unsigned","value":5}]]}]})
28}
29fn send() -> ActionDefinition {
30 ActionDefinition{name:"export_ipfix_records".into(),description:"Validate one complete typed UDP message before emitting. Each batch repeats all referenced templates; domain sequence advances by locally sent data/options records. No acknowledgment or automatic data retry.".into(),parameters:vec![parameter("batch","object","observation_domain_id;optional unsigned32 export_time;1..32 templates(id>=256,optional scope_count,ordered fields with supported element and optional length);ordered data_sets(template_id,records of aligned typed field values). Kinds unsigned/ipv4/ipv6/string/timestamp_seconds/timestamp_milliseconds. Bounds8192bytes,64sets,256records,32fields,1024byte strings; reduced unsigned sizes1..native width; strings fixed1..1024 or variable65535.",true)],example:json!({"type":"export_ipfix_records","batch":example_batch()}),log_template:Some(LogTemplate::new().with_info("Export a validated IPFIX batch"))}
31}
32fn disconnect() -> ActionDefinition {
33 ActionDefinition {
34 name: "disconnect".into(),
35 description: "Cancel UDP sends, template refresh, handlers and command handle".into(),
36 parameters: vec![],
37 example: json!({"type":"disconnect"}),
38 log_template: Some(LogTemplate::new().with_info("Disconnect IPFIX exporter")),
39 }
40}
41pub static IPFIX_CONNECTED_EVENT: LazyLock<EventType> = LazyLock::new(|| {
42 EventType::new(
43 "ipfix_connected",
44 "UDP exporter ready; collector reception is unconfirmed.",
45 send().example,
46 )
47 .with_parameters(vec![
48 parameter(
49 "remote_addr",
50 "string",
51 "Remote IPFIX collector socket address",
52 true,
53 ),
54 parameter("local_addr", "string", "Exporter UDP socket", true),
55 ])
56 .with_actions(vec![send(), disconnect()])
57});
58pub static IPFIX_EXPORTED_EVENT: LazyLock<EventType> = LazyLock::new(|| {
59 EventType::new("ipfix_exported","One datagram accepted by the local UDP transport, not an end-to-end receipt or persistence acknowledgment.",send().example).with_parameters(vec![parameter("observation_domain_id","number","Exporter observation domain identifier",true),parameter("sequence_number","number","Header sequence before increment",true),parameter("record_count","number","Data/options records",true),parameter("template_count","number","Number of templates in this datagram",true),parameter("byte_count","number","Datagram length",true),parameter("local_transport_only","boolean","True when the local UDP transport accepted this datagram",true)]).with_actions(vec![send(),disconnect()])
60});
61impl Protocol for IpfixClientProtocol {
62 fn protocol_name(&self) -> &'static str {
63 "IPFIX"
64 }
65 fn stack_name(&self) -> &'static str {
66 "ETH>IP>UDP>IPFIX"
67 }
68 fn keywords(&self) -> Vec<&'static str> {
69 vec!["ipfix", "ipfix-udp"]
70 }
71 fn description(&self) -> &'static str {
72 "Typed UDP IPFIX exporter with template refresh and per-domain sequence"
73 }
74 fn example_prompt(&self) -> &'static str {
75 "Export IPFIX records to localhost:4739"
76 }
77 fn group_name(&self) -> &'static str {
78 "Network Management"
79 }
80 fn get_async_actions(&self, _: &AppState) -> Vec<ActionDefinition> {
81 vec![send(), disconnect()]
82 }
83 fn get_sync_actions(&self) -> Vec<ActionDefinition> {
84 vec![]
85 }
86 fn get_event_types(&self) -> Vec<EventType> {
87 vec![IPFIX_CONNECTED_EVENT.clone(), IPFIX_EXPORTED_EVENT.clone()]
88 }
89 fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
90 vec![duration_parameter("template_refresh_seconds",super::transport::DEFAULT_REFRESH_SECONDS,"Periodic active-template retransmission,1..3600seconds; every record batch also repeats its templates")]
91 }
92 fn metadata(&self) -> ProtocolMetadataV2 {
93 ProtocolMetadataV2::builder().state(DevelopmentState::Experimental).well_known_udp_port(4739).implementation("Native typed RFC7011 UDP template/options/data encoder; owned periodic template retransmission").llm_control("Connected/local-export events with standard shared memory, bounded followups and injection independent of parked handlers").e2e_testing("Pinned independent Python IPFIX0.9.7 decoder plus live official GoFlow2 2.2.7 collector; native pair, malformed/atomic bounds and cancellation").notes("UDP-only Experimental subset; no SCTP/TCP/TLS/DTLS, authentication, full information-model, flow store, ACK, durability or data retry.8192bytes/message,64sets,256records,32fields,1024byte strings;32domains,32templates/domain. All referenced templates supplied with every batch; identical definitions refresh and template ID redefinition rejected until a new logical socket. Periodic template-only sends default60s, configurable1..3600s; domain sequence counts locally emitted data/options records modulo2^32.10s resolve/send deadline, one in-flight send;32queued events/actions, followup depth8. Transport acceptance is not collector reception. No fuzz/pcap claim.").build()
94 }
95 fn get_startup_examples(&self) -> StartupExamples {
96 StartupExamples::new(
97 json!({"type":"open_client","base_stack":"ipfix","remote_addr":"localhost:4739","instruction":"Export a flow record"}),
98 json!({"type":"open_client","base_stack":"ipfix","remote_addr":"localhost:4739","event_handlers":[{"event_pattern":"ipfix_connected","handler":{"type":"script","language":"python","code":format!("import json,sys\njson.load(sys.stdin)\nprint({:?})",json!({"actions":[send().example]}).to_string())}}]}),
99 json!({"type":"open_client","base_stack":"ipfix","remote_addr":"localhost:4739","event_handlers":[{"event_pattern":"ipfix_connected","handler":{"type":"static","actions":[send().example]}}]}),
100 )
101 }
102}
103impl Client for IpfixClientProtocol {
104 fn connect(
105 &self,
106 ctx: ConnectContext,
107 ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
108 {
109 Box::pin(super::IpfixClient::connect(ctx))
110 }
111 fn execute_action(&self, action: Value) -> Result<ClientActionResult> {
112 match action["type"].as_str() {
113 Some("export_ipfix_records") => {
114 let batch: Batch =
115 serde_json::from_value(action.get("batch").context("missing batch")?.clone())?;
116 codec::encode(&batch, 0, batch.export_time.unwrap_or(0))?;
117 Ok(ClientActionResult::Custom {
118 name: "export_ipfix_records".into(),
119 data: serde_json::to_value(batch)?,
120 })
121 }
122 Some("disconnect") => Ok(ClientActionResult::Disconnect),
123 _ => bail!("unknown IPFIX exporter action"),
124 }
125 }
126}