netget/server/ipfix/
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 IpfixProtocol;
19impl IpfixProtocol {
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_ipfix_records".into(),description:"Observe validated typed IPFIX records in the bounded common access log. UDP has no acknowledgment; no persistence or aggregation.".into(),parameters:vec![],example:json!({"type":"collect_ipfix_records"}),log_template:Some(LogTemplate::new().with_info("Collect validated IPFIX records"))}
34}
35pub static IPFIX_MESSAGE_EVENT: LazyLock<EventType> = LazyLock::new(|| {
36    EventType::new("ipfix_message","One validated UDP IPFIX message. Ordered field values align with each set's template; unsupported IEs 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, ignored UDP withdrawals, 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 IpfixProtocol {
49    fn protocol_name(&self) -> &'static str {
50        "IPFIX"
51    }
52    fn stack_name(&self) -> &'static str {
53        "ETH>IP>UDP>IPFIX"
54    }
55    fn keywords(&self) -> Vec<&'static str> {
56        vec!["ipfix", "ipfix-udp"]
57    }
58    fn description(&self) -> &'static str {
59        "Typed bounded UDP IPFIX templates, options and records; no model calls by default"
60    }
61    fn example_prompt(&self) -> &'static str {
62        "Collect IPFIX records on UDP4739"
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![IPFIX_MESSAGE_EVENT.clone()]
75    }
76    fn metadata(&self) -> ProtocolMetadataV2 {
77        ProtocolMetadataV2::builder().deliberately_silent().state(DevelopmentState::Experimental).well_known_udp_port(4739).max_inbound_bytes(MAX_MESSAGE_BYTES).implementation("Native bounded RFC7011 UDP framing and transient template/sequence state; typed IANA subset, no protocol library dependency").llm_control("Ordered typed records through common handlers/shared memory; llm_fallback=false collects unmatched messages without model calls").e2e_testing("Independent pinned Python IPFIX0.9.7 exporter/decoder and official GoFlow2 2.2.7 collector; native pair, malformed/bounds, template/sequence/lifecycle checks").notes("Experimental UDP subset, not full RFC7011 compliance: no mandatory SCTP, TCP/TLS/DTLS, authentication, retransmission recovery, flow store, persistence or aggregation.8192byte messages,64sets,256records,32fields,1024byte scalar/string/unknown fields;128peer/domain sessions,32templates/session,1024global templates;32queued handler events. Configurable template TTL(default600s) and session idle(default1800s), each1..86400s. Unknown/enterprise values discarded with descriptors; unknown template sets counted but not buffered and sequence expectation becomes untracked. UDP withdrawals ignored; identical refresh extends TTL without a template-change report; redefinitions replace in wire order; malformed messages do not partially mutate template/sequence state. Sequence modulo2^32 counts data/options records; out-of-order/duplicate observations do not regress expected sequence. No ACK/error replies; failed handlers are logged and cannot roll back prior wire-state parsing. No fuzz/pcap/full information-model claim.").build()
78    }
79    fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
80        vec![
81            duration_parameter(
82                "template_ttl_seconds",
83                DEFAULT_TEMPLATE_TTL,
84                "Template lifetime since refresh,1..86400seconds; a wire cache only",
85            ),
86            duration_parameter(
87                "session_idle_seconds",
88                DEFAULT_SESSION_IDLE,
89                "Peer/domain idle state expiry,1..86400seconds",
90            ),
91            ParameterDefinition {
92                name: "llm_fallback".into(),
93                type_hint: "boolean".into(),
94                description:
95                    "Opt unmatched messages into model calls; explicit handlers always run".into(),
96                required: false,
97                example: json!(true),
98                default: Some(json!(DEFAULT_LLM_FALLBACK)),
99            },
100        ]
101    }
102    fn get_startup_examples(&self) -> StartupExamples {
103        StartupExamples::new(
104            json!({"type":"open_server","base_stack":"ipfix","port":4739,"startup_params":{"llm_fallback":true},"instruction":"Inspect flow records"}),
105            json!({"type":"open_server","base_stack":"ipfix","port":4739,"event_handlers":[{"event_pattern":"ipfix_message","handler":{"type":"script","language":"python","code":"import json,sys\njson.load(sys.stdin)\nprint(json.dumps({'actions':[{'type':'collect_ipfix_records'}]}))"}}]}),
106            json!({"type":"open_server","base_stack":"ipfix","port":4739,"event_handlers":[{"event_pattern":"ipfix_message","handler":{"type":"static","actions":[collect().example]}}]}),
107        )
108    }
109}
110impl Server for IpfixProtocol {
111    fn spawn(
112        &self,
113        ctx: SpawnContext,
114    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
115    {
116        Box::pin(super::IpfixServer::spawn(ctx))
117    }
118    fn execute_action(&self, action: Value) -> Result<ActionResult> {
119        match action["type"].as_str() {
120            Some("collect_ipfix_records") => Ok(ActionResult::NoAction),
121            _ => bail!("unknown IPFIX collector action"),
122        }
123    }
124}
netget/server/ipfix/actions.rs

netget/server/ipfix/
actions.rs
