netget/server/netflow_v9/
actions.rs

1use super::codec::{
2    DEFAULT_LLM_FALLBACK, DEFAULT_SESSION_IDLE, DEFAULT_TEMPLATE_TTL, MAX_MESSAGE_BYTES,
3};
4use crate::llm::actions::{
5    protocol_trait::{ActionResult, Protocol, Server},
6    ActionDefinition, Parameter, ParameterDefinition, StartupExamples,
7};
8use crate::protocol::log_template::LogTemplate;
9use crate::protocol::{
10    metadata::{DevelopmentState, ProtocolMetadataV2},
11    EventType, SpawnContext,
12};
13use crate::state::app_state::AppState;
14use anyhow::{bail, Result};
15use serde_json::{json, Value};
16use std::sync::LazyLock;
17#[derive(Default)]
18pub struct NetflowV9Protocol;
19impl NetflowV9Protocol {
20    pub fn new() -> Self {
21        Self
22    }
23}
24pub fn parameter(name: &str, type_hint: &str, description: &str, required: bool) -> Parameter {
25    Parameter {
26        name: name.into(),
27        type_hint: type_hint.into(),
28        description: description.into(),
29        required,
30    }
31}
32fn collect() -> ActionDefinition {
33    ActionDefinition{name:"collect_netflow_v9_records".into(),description:"Observe validated typed NETFLOW_V9 records in the bounded common access log. UDP has no acknowledgment; no persistence or aggregation.".into(),parameters:vec![],example:json!({"type":"collect_netflow_v9_records"}),log_template:Some(LogTemplate::new().with_info("NetFlow v9 records observed without acknowledgment"))}
34}
35pub static NETFLOW_V9_MESSAGE_EVENT: LazyLock<EventType> = LazyLock::new(|| {
36    EventType::new("netflow_v9_message","One validated UDP NETFLOW_V9 message. Ordered field values align with each set's template; unsupported v9 fields retain descriptors and null values without raw bytes. Sequence gaps are observations, not recovery.",collect().example).with_parameters(vec![parameter("message","object","Export time, domain, sequence/tracking, template changes, header total record Count/status, sysUpTime, typed data sets, unknown-set descriptors and record count. Scope fields are the first scope_count ordered fields; duplicate IE occurrences preserve order.",true),parameter("source_addr","string","Exporter IP and UDP source port, scoped to this collector socket",true)]).with_actions(vec![collect()])
37});
38pub fn duration_parameter(name: &str, default: u64, description: &str) -> ParameterDefinition {
39    ParameterDefinition {
40        name: name.into(),
41        type_hint: "number".into(),
42        description: description.into(),
43        required: false,
44        example: json!(default),
45        default: Some(json!(default)),
46    }
47}
48impl Protocol for NetflowV9Protocol {
49    fn protocol_name(&self) -> &'static str {
50        "NetFlowV9"
51    }
52    fn stack_name(&self) -> &'static str {
53        "ETH>IP>UDP>NetFlowV9"
54    }
55    fn keywords(&self) -> Vec<&'static str> {
56        vec!["netflow-v9", "netflowv9", "netflow_v9"]
57    }
58    fn description(&self) -> &'static str {
59        "Typed bounded UDP NetFlow v9 templates, options and records; no model calls by default"
60    }
61    fn example_prompt(&self) -> &'static str {
62        "Collect NETFLOW_V9 records on UDP2055"
63    }
64    fn group_name(&self) -> &'static str {
65        "Network Management"
66    }
67    fn get_async_actions(&self, _: &AppState) -> Vec<ActionDefinition> {
68        vec![]
69    }
70    fn get_sync_actions(&self) -> Vec<ActionDefinition> {
71        vec![collect()]
72    }
73    fn get_event_types(&self) -> Vec<EventType> {
74        vec![NETFLOW_V9_MESSAGE_EVENT.clone()]
75    }
76    fn metadata(&self) -> ProtocolMetadataV2 {
77        ProtocolMetadataV2::builder().deliberately_silent().request_only("One-way UDP collector: no replies or unsolicited messages").state(DevelopmentState::Experimental).well_known_udp_port(2055).max_inbound_bytes(MAX_MESSAGE_BYTES)
78        .implementation("Native RFC3954 version9 UDP header, normal/options templates, fixed typed fields and transient cache; no linked protocol dependency")
79        .llm_control("Ordered typed records through common handlers/shared memory; llm_fallback=false observes unmatched datagrams without model calls")
80        .e2e_testing("Literal header/count/scopes both directions; required unmodified softflowd1.1.1 upstream legacy exporter and actual GoFlow2 2.2.7 collector; atomic bounds, source/sequence, expiry, silent failures and owned cancellation")
81        .notes("Experimental bounded UDP subset, not full RFC3954 compliance.20byte header Count includes all normal/options template and data/options records; sequence counts export packets modulo2^32 including template-only packets. Source-IP+Source-ID cache key intentionally ignores UDP port.8192bytes,64flowsets,32templates/message,256data records,32fields,1..1024byte fixed fields,minimum record4bytes to avoid ambiguous padding;128sessions,32templates/session,1024global templates,32queued events.35 selected RFC scalar fields plus5 unsigned options scopes; unsupported field values discarded as null with full16bit descriptors, no IPFIX enterprise bit or variable-length encoding. Unknown-template data not buffered; Count then marked unverifiable but packet tracking continues. Known-record Count mismatch and malformed packet fail atomically. Identical templates refresh TTL600s, differing templates replace in arrival order; idle1800s; both1..86400s. A backwards Unix export clock on an advancing packet clears candidate templates; lower uptime only flagged due wrap/restart/reordering ambiguity. No withdrawal format, ACK/error replies, authentication, TCP/SCTP/TLS, full vendor model, flow storage, aggregation, durability, delayed-data recovery, fuzz or capture claim. Common handler failure cannot undo committed wire parsing.").build()
82    }
83    fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
84        vec![
85            duration_parameter(
86                "template_ttl_seconds",
87                DEFAULT_TEMPLATE_TTL,
88                "Template lifetime since refresh,1..86400seconds; a wire cache only",
89            ),
90            duration_parameter(
91                "session_idle_seconds",
92                DEFAULT_SESSION_IDLE,
93                "Peer/domain idle state expiry,1..86400seconds",
94            ),
95            ParameterDefinition {
96                name: "llm_fallback".into(),
97                type_hint: "boolean".into(),
98                description:
99                    "Opt unmatched messages into model calls; explicit handlers always run".into(),
100                required: false,
101                example: json!(true),
102                default: Some(json!(DEFAULT_LLM_FALLBACK)),
103            },
104        ]
105    }
106    fn get_startup_examples(&self) -> StartupExamples {
107        StartupExamples::new(
108            json!({"type":"open_server","base_stack":"netflow_v9","port":2055,"startup_params":{"llm_fallback":true},"instruction":"Inspect flow records"}),
109            json!({"type":"open_server","base_stack":"netflow_v9","port":2055,"event_handlers":[{"event_pattern":"netflow_v9_message","handler":{"type":"script","language":"python","code":"import json,sys\njson.load(sys.stdin)\nprint(json.dumps({'actions':[{'type':'collect_netflow_v9_records'}]}))"}}]}),
110            json!({"type":"open_server","base_stack":"netflow_v9","port":2055,"event_handlers":[{"event_pattern":"netflow_v9_message","handler":{"type":"static","actions":[collect().example]}}]}),
111        )
112    }
113}
114impl Server for NetflowV9Protocol {
115    fn spawn(
116        &self,
117        ctx: SpawnContext,
118    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
119    {
120        Box::pin(super::NetflowV9Server::spawn(ctx))
121    }
122    fn execute_action(&self, action: Value) -> Result<ActionResult> {
123        match action["type"].as_str() {
124            Some("collect_netflow_v9_records") => Ok(ActionResult::NoAction),
125            _ => bail!("unknown NETFLOW_V9 collector action"),
126        }
127    }
128}
netget/server/netflow_v9/actions.rs

netget/server/netflow_v9/
actions.rs
