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::netflow_v9::{
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 NetflowV9ClientProtocol;
21impl NetflowV9ClientProtocol {
22 pub fn new() -> Self {
23 Self
24 }
25}
26pub fn example_batch() -> Value {
27 json!({"source_id":42,"templates":[{"id":256,"fields":[{"element":"source_ipv4"},{"element":"destination_ipv4"},{"element":"in_packets"}]}],"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_netflow_v9_records".into(),description:"Validate one complete typed v9 UDP packet before emitting; repeats every referenced template, counts all records in Count and advances Source-ID sequence by one packet. Local transport only, no ACK or data retry.".into(),parameters:vec![parameter("batch","object","source_id:u32;optional export_time:u32,sys_uptime_ms:u32;1..32 templates(id>=256,scope_count default0,ordered fields declaring exactly one supported element or scope and optional fixed length);data_sets(template_id,records of aligned {kind,value}). Kinds unsigned/ipv4/ipv6/uptime_milliseconds. Options scopes first(system/interface/line_card/cache/template), followed by options.8192bytes,64flowsets,256data records,32fields,record size>=4; no variable fields or raw bytes.",true)],example:json!({"type":"export_netflow_v9_records","batch":example_batch()}),log_template:Some(LogTemplate::new().with_info("NetFlow v9 export queued for Source ID {batch.source_id}"))}
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("NetFlow v9 exporter disconnected")),
39 }
40}
41pub static NETFLOW_V9_CONNECTED_EVENT: LazyLock<EventType> = LazyLock::new(|| {
42 EventType::new(
43 "netflow_v9_connected",
44 "UDP exporter ready; collector reception is unconfirmed.",
45 send().example,
46 )
47 .with_parameters(vec![
48 parameter("remote_addr", "string", "Collector UDP address", true),
49 parameter("local_addr", "string", "Exporter UDP socket", true),
50 ])
51 .with_actions(vec![send(), disconnect()])
52});
53pub static NETFLOW_V9_EXPORTED_EVENT: LazyLock<EventType> = LazyLock::new(|| {
54 EventType::new("netflow_v9_exported","One datagram accepted by the local UDP transport, not an end-to-end receipt or persistence acknowledgment.",send().example).with_parameters(vec![parameter("source_id","number","Observation Source ID",true),parameter("sequence_number","number","Header sequence before increment",true),parameter("record_count","number","Data/options records",true),parameter("header_count","number","All template and data/options records",true),parameter("sys_uptime_ms","number","Export uptime milliseconds",true),parameter("template_count","number","Templates supplied in this packet",true),parameter("byte_count","number","Datagram length",true),parameter("local_transport_only","boolean","Always true; local UDP send does not confirm reception",true)]).with_actions(vec![send(),disconnect()])
55});
56impl Protocol for NetflowV9ClientProtocol {
57 fn protocol_name(&self) -> &'static str {
58 "NetFlowV9"
59 }
60 fn stack_name(&self) -> &'static str {
61 "ETH>IP>UDP>NetFlowV9"
62 }
63 fn keywords(&self) -> Vec<&'static str> {
64 vec!["netflow-v9", "netflowv9", "netflow_v9"]
65 }
66 fn description(&self) -> &'static str {
67 "Typed UDP NetFlow v9 exporter with template refresh and per-Source-ID packet sequence"
68 }
69 fn example_prompt(&self) -> &'static str {
70 "Export NETFLOW_V9 records to localhost:2055"
71 }
72 fn group_name(&self) -> &'static str {
73 "Network Management"
74 }
75 fn get_async_actions(&self, _: &AppState) -> Vec<ActionDefinition> {
76 vec![send(), disconnect()]
77 }
78 fn get_sync_actions(&self) -> Vec<ActionDefinition> {
79 vec![]
80 }
81 fn get_event_types(&self) -> Vec<EventType> {
82 vec![
83 NETFLOW_V9_CONNECTED_EVENT.clone(),
84 NETFLOW_V9_EXPORTED_EVENT.clone(),
85 ]
86 }
87 fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
88 vec![duration_parameter("template_refresh_seconds",super::transport::DEFAULT_REFRESH_SECONDS,"Periodic active-template retransmission,1..3600seconds; every record batch also repeats its templates")]
89 }
90 fn metadata(&self) -> ProtocolMetadataV2 {
91 ProtocolMetadataV2::builder().state(DevelopmentState::Experimental).well_known_udp_port(2055)
92 .implementation("Native typed RFC3954 v9 UDP normal/options template and data encoder; owned periodic template refresh")
93 .llm_control("Connected/local-export events through common handlers/shared memory; bounded actions and injection independent of parked handlers")
94 .e2e_testing("Required actual GoFlow2 2.2.7 service independently decodes exported IPv4/IPv6, counters, uptime and options scopes; native literal/lifecycle/atomic bounds and unmodified softflowd exporter cross-check")
95 .notes("Experimental UDP-only selected RFC3954 subset. Count includes templates and data/options records; Source-ID sequence advances once per locally sent packet, including template-only refresh, modulo2^32.32Source IDs,32active templates each; identical definitions allowed, ID redefinition rejected until a new exporter instance. All referenced templates supplied per batch. Periodic time refresh60s configurable1..3600s; no packet-count refresh knob or data retry. sysUpTime defaults to per-source monotonic elapsed time since its first batch; explicit unsigned32 override carried forward for refresh. Unix seconds default system clock or caller unsigned32 override.8192bytes,64flowsets,256data records,32fields,minimum record4bytes;35selected RFC fields,5unsigned scopes, fixed-width data only. One in-flight operation,10sresolve/send,32events/actions,depth8. Refresh candidates commit only after all local sends; failure closes and discards transient state. UDP acceptance does not prove receipt/processing/storage. No ACK, TCP/SCTP/TLS/auth, variable/enterprise extensions, full vendor model, flow capture/store, durability, fuzz or packet-capture evidence.").build()
96 }
97 fn get_startup_examples(&self) -> StartupExamples {
98 StartupExamples::new(
99 json!({"type":"open_client","base_stack":"netflow_v9","remote_addr":"localhost:2055","instruction":"Export a flow record"}),
100 json!({"type":"open_client","base_stack":"netflow_v9","remote_addr":"localhost:2055","event_handlers":[{"event_pattern":"netflow_v9_connected","handler":{"type":"script","language":"python","code":format!("import json,sys\njson.load(sys.stdin)\nprint({:?})",json!({"actions":[send().example]}).to_string())}}]}),
101 json!({"type":"open_client","base_stack":"netflow_v9","remote_addr":"localhost:2055","event_handlers":[{"event_pattern":"netflow_v9_connected","handler":{"type":"static","actions":[send().example]}}]}),
102 )
103 }
104}
105impl Client for NetflowV9ClientProtocol {
106 fn connect(
107 &self,
108 ctx: ConnectContext,
109 ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
110 {
111 Box::pin(super::NetflowV9Client::connect(ctx))
112 }
113 fn execute_action(&self, action: Value) -> Result<ClientActionResult> {
114 match action["type"].as_str() {
115 Some("export_netflow_v9_records") => {
116 let batch: Batch =
117 serde_json::from_value(action.get("batch").context("missing batch")?.clone())?;
118 codec::encode(
119 &batch,
120 0,
121 batch.export_time.unwrap_or(0),
122 batch.sys_uptime_ms.unwrap_or(0),
123 )?;
124 Ok(ClientActionResult::Custom {
125 name: "export_netflow_v9_records".into(),
126 data: serde_json::to_value(batch)?,
127 })
128 }
129 Some("disconnect") => Ok(ClientActionResult::Disconnect),
130 _ => bail!("unknown NETFLOW_V9 exporter action"),
131 }
132 }
133}