async_snmp/agent/
notification.rs

1//! Agent notification sending (trap/inform).
2//!
3//! Provides trap sink configuration and methods for sending notifications
4//! from an agent to configured destinations.
5
6use std::net::SocketAddr;
7use std::sync::RwLock;
8use std::time::Duration;
9
10use bytes::Bytes;
11use tokio::sync::Mutex as AsyncMutex;
12
13use crate::client::{Auth, Client, ClientConfig, CommunityVersion, Retry};
14use crate::error::{Error, Result};
15use crate::message::CommunityMessage;
16use crate::oid::Oid;
17use crate::pdu::Pdu;
18use crate::transport::{UdpHandle, UdpTransport};
19use crate::v3::{DerivedKeys, UsmConfig};
20use crate::varbind::VarBind;
21use crate::version::Version;
22
23/// A configured notification destination.
24///
25/// Stores resolved credentials and cached keys for sending traps and informs
26/// to a specific target.
27pub(crate) struct TrapSink {
28    pub(crate) dest: SocketAddr,
29    pub(crate) version: Version,
30    pub(crate) community: Bytes,
31    pub(crate) v3_security: Option<UsmConfig>,
32    /// Keys derived against the agent's `engine_id` for V3 trap sending.
33    /// Lazily populated on first use.
34    pub(crate) derived_keys: RwLock<Option<DerivedKeys>>,
35    /// Inform request timeout and retry policy.
36    inform_timeout: Duration,
37    inform_retry: Retry,
38    /// Cached client for inform sending. Lazily created on first inform.
39    /// Holds both the transport (to keep the socket alive) and the client.
40    inform_client: AsyncMutex<Option<(UdpTransport, Client<UdpHandle>)>>,
41}
42
43impl TrapSink {
44    /// Create from an Auth configuration and resolved destination address.
45    pub(crate) fn new(
46        dest: SocketAddr,
47        auth: Auth,
48        inform_timeout: Duration,
49        inform_retry: Retry,
50    ) -> Self {
51        match auth {
52            Auth::Community { version, community } => {
53                let snmp_version = match version {
54                    CommunityVersion::V1 => Version::V1,
55                    CommunityVersion::V2c => Version::V2c,
56                };
57                TrapSink {
58                    dest,
59                    version: snmp_version,
60                    community: Bytes::copy_from_slice(community.as_bytes()),
61                    v3_security: None,
62                    derived_keys: RwLock::new(None),
63                    inform_timeout,
64                    inform_retry,
65                    inform_client: AsyncMutex::new(None),
66                }
67            }
68            Auth::Usm(security) => TrapSink {
69                dest,
70                version: Version::V3,
71                community: Bytes::new(),
72                v3_security: Some(security),
73                derived_keys: RwLock::new(None),
74                inform_timeout,
75                inform_retry,
76                inform_client: AsyncMutex::new(None),
77            },
78        }
79    }
80
81    /// Ensure keys are derived against the given `engine_id` for V3 trap sending.
82    fn ensure_keys_derived(&self, engine_id: &[u8]) -> Result<()> {
83        {
84            let keys = self.derived_keys.read().map_err(|_| {
85                Error::Config("trap sink derived_keys lock poisoned".into()).boxed()
86            })?;
87            if keys.is_some() {
88                return Ok(());
89            }
90        }
91
92        let security = self.v3_security.as_ref().ok_or_else(|| {
93            Error::Config("V3 security not configured for trap sink".into()).boxed()
94        })?;
95
96        let keys = security
97            .derive_keys(engine_id)
98            .map_err(|e| Error::Config(e.to_string().into()).boxed())?;
99
100        let mut derived = self
101            .derived_keys
102            .write()
103            .map_err(|_| Error::Config("trap sink derived_keys lock poisoned".into()).boxed())?;
104        *derived = Some(keys);
105
106        Ok(())
107    }
108
109    /// Get or create the cached inform client for this sink.
110    async fn get_or_create_inform_client(&self) -> Result<Client<UdpHandle>> {
111        let mut guard = self.inform_client.lock().await;
112        if let Some((_, ref client)) = *guard {
113            return Ok(client.clone());
114        }
115
116        let config = match self.version {
117            Version::V1 => unreachable!("v1 does not support informs"),
118            Version::V2c => ClientConfig {
119                version: Version::V2c,
120                community: self.community.clone(),
121                timeout: self.inform_timeout,
122                retry: self.inform_retry.clone(),
123                v3_security: None,
124                ..ClientConfig::default()
125            },
126            Version::V3 => ClientConfig {
127                version: Version::V3,
128                community: Bytes::new(),
129                timeout: self.inform_timeout,
130                retry: self.inform_retry.clone(),
131                v3_security: self.v3_security.clone(),
132                ..ClientConfig::default()
133            },
134        };
135
136        let bind_addr = if self.dest.is_ipv6() {
137            "[::]:0"
138        } else {
139            "0.0.0.0:0"
140        };
141        let transport = UdpTransport::bind(bind_addr).await?;
142        let handle = transport.handle(self.dest);
143        let client = Client::new(handle, config);
144        *guard = Some((transport, client.clone()));
145        Ok(client)
146    }
147}
148
149/// Delivery outcome for a single trap sink.
150///
151/// Reports the destination and the result of the delivery attempt. For traps
152/// this reflects the local send (encoding and socket write); for confirmed
153/// informs it reflects the full request/response exchange, including timeout.
154#[derive(Debug)]
155pub struct SinkOutcome {
156    /// The sink destination address.
157    pub dest: SocketAddr,
158    /// The delivery result for this sink. `Ok(())` on success.
159    pub result: Result<()>,
160}
161
162/// Aggregate outcome of sending a notification to all configured sinks.
163///
164/// Returned by [`Agent::send_trap_detailed`](super::Agent::send_trap_detailed)
165/// and [`Agent::send_inform_detailed`](super::Agent::send_inform_detailed) so
166/// callers can observe partial success: which sinks succeeded and which failed
167/// with their errors. Sinks that were skipped (e.g. v1 sinks for informs) are
168/// not included.
169#[derive(Debug)]
170pub struct NotificationOutcome {
171    sinks: Vec<SinkOutcome>,
172}
173
174impl NotificationOutcome {
175    /// Per-sink outcomes, in sink configuration order.
176    pub fn sinks(&self) -> &[SinkOutcome] {
177        &self.sinks
178    }
179
180    /// Iterator over the sinks whose delivery failed.
181    pub fn failures(&self) -> impl Iterator<Item = &SinkOutcome> {
182        self.sinks.iter().filter(|s| s.result.is_err())
183    }
184
185    /// `true` if every attempted sink succeeded (also `true` when no sinks
186    /// were attempted).
187    pub fn all_succeeded(&self) -> bool {
188        self.sinks.iter().all(|s| s.result.is_ok())
189    }
190
191    /// Number of sinks attempted.
192    pub fn len(&self) -> usize {
193        self.sinks.len()
194    }
195
196    /// `true` if no sinks were attempted.
197    pub fn is_empty(&self) -> bool {
198        self.sinks.is_empty()
199    }
200
201    /// Consume the outcome, returning the per-sink outcomes.
202    pub fn into_sinks(self) -> Vec<SinkOutcome> {
203        self.sinks
204    }
205}
206
207impl super::Agent {
208    /// Send a trap to all configured trap sinks.
209    ///
210    /// Constructs a `TrapV2` PDU with the mandatory sysUpTime.0 and snmpTrapOID.0
211    /// prefix and sends it to each destination. Fire-and-forget: no response
212    /// expected.
213    ///
214    /// V1 trap sinks receive a converted v1 trap (RFC 3584 Section 3.2).
215    /// For a V3 sink, this Agent is authoritative and sends its persisted
216    /// engine ID with the current boots/time tuple.
217    ///
218    /// # Example
219    ///
220    /// ```rust,no_run
221    /// # use async_snmp::agent::Agent;
222    /// # use async_snmp::{Auth, oid};
223    /// # async fn example() -> Result<(), Box<async_snmp::Error>> {
224    /// let agent = Agent::builder()
225    ///     .bind("0.0.0.0:1161")
226    ///     .community(b"public")
227    ///     .trap_sink("192.168.1.100:162", Auth::v2c("public"))
228    ///     .build()
229    ///     .await?;
230    ///
231    /// // Send coldStart trap to all sinks
232    /// agent.send_trap(&oid!(1, 3, 6, 1, 6, 3, 1, 1, 5, 1), 0, vec![]).await?;
233    /// # Ok(())
234    /// # }
235    /// ```
236    pub async fn send_trap(
237        &self,
238        trap_oid: &Oid,
239        uptime: u32,
240        varbinds: Vec<VarBind>,
241    ) -> Result<()> {
242        let outcome = self.send_trap_detailed(trap_oid, uptime, varbinds).await;
243        for sink in outcome.failures() {
244            if let Err(ref e) = sink.result {
245                tracing::warn!(target: "async_snmp::agent", { snmp.dest = %sink.dest, error = %e }, "failed to send trap");
246            }
247        }
248        Ok(())
249    }
250
251    /// Send a trap to all configured trap sinks, reporting per-sink outcomes.
252    ///
253    /// Behaves like [`send_trap`](Self::send_trap) but returns a
254    /// [`NotificationOutcome`] describing which sinks succeeded and which
255    /// failed with their errors, rather than discarding per-sink failures.
256    /// Unlike [`send_trap`](Self::send_trap), it does not emit warning logs for
257    /// failures; the caller is expected to inspect the returned outcome.
258    pub async fn send_trap_detailed(
259        &self,
260        trap_oid: &Oid,
261        uptime: u32,
262        varbinds: Vec<VarBind>,
263    ) -> NotificationOutcome {
264        let sinks = &self.inner.trap_sinks;
265        let mut outcomes = Vec::with_capacity(sinks.len());
266        if sinks.is_empty() {
267            return NotificationOutcome { sinks: outcomes };
268        }
269
270        let request_id = self.next_notification_id();
271        let pdu = Pdu::trap_v2(request_id, uptime, trap_oid, varbinds);
272
273        for sink in sinks {
274            let result = self.send_trap_to_sink(sink, &pdu).await;
275            outcomes.push(SinkOutcome {
276                dest: sink.dest,
277                result,
278            });
279        }
280
281        NotificationOutcome { sinks: outcomes }
282    }
283
284    /// Send an inform to all configured trap sinks.
285    ///
286    /// Constructs an `InformRequest` PDU and sends it to each destination,
287    /// waiting for acknowledgement from each. Reuses a cached client per
288    /// sink for the request/response exchange.
289    ///
290    /// V1 trap sinks are skipped (v1 does not support informs).
291    /// For a V3 sink, the receiver is authoritative; the cached client
292    /// discovers and uses the sink's engine identity and trusted time rather
293    /// than the Agent's local authoritative state.
294    ///
295    /// # Example
296    ///
297    /// ```rust,no_run
298    /// # use async_snmp::agent::Agent;
299    /// # use async_snmp::{Auth, oid};
300    /// # async fn example() -> Result<(), Box<async_snmp::Error>> {
301    /// let agent = Agent::builder()
302    ///     .bind("0.0.0.0:1161")
303    ///     .community(b"public")
304    ///     .trap_sink("192.168.1.100:162", Auth::v2c("public"))
305    ///     .build()
306    ///     .await?;
307    ///
308    /// // Send warmStart inform to all sinks (waits for acknowledgement)
309    /// agent.send_inform(&oid!(1, 3, 6, 1, 6, 3, 1, 1, 5, 2), 0, vec![]).await?;
310    /// # Ok(())
311    /// # }
312    /// ```
313    pub async fn send_inform(
314        &self,
315        trap_oid: &Oid,
316        uptime: u32,
317        varbinds: Vec<VarBind>,
318    ) -> Result<()> {
319        let outcome = self.send_inform_detailed(trap_oid, uptime, varbinds).await;
320        for sink in outcome.failures() {
321            if let Err(ref e) = sink.result {
322                tracing::warn!(target: "async_snmp::agent", { snmp.dest = %sink.dest, error = %e }, "failed to send inform");
323            }
324        }
325        Ok(())
326    }
327
328    /// Send an inform to all configured trap sinks, reporting per-sink outcomes.
329    ///
330    /// Behaves like [`send_inform`](Self::send_inform) but returns a
331    /// [`NotificationOutcome`] describing which sinks acknowledged and which
332    /// failed with their errors (including confirmed-inform timeouts), rather
333    /// than discarding per-sink failures. v1 sinks are skipped and are not
334    /// included in the outcome. Unlike [`send_inform`](Self::send_inform), it
335    /// does not emit warning logs for failures; the caller is expected to
336    /// inspect the returned outcome.
337    pub async fn send_inform_detailed(
338        &self,
339        trap_oid: &Oid,
340        uptime: u32,
341        varbinds: Vec<VarBind>,
342    ) -> NotificationOutcome {
343        let sinks = &self.inner.trap_sinks;
344        let mut outcomes = Vec::new();
345
346        for sink in sinks {
347            if sink.version == Version::V1 {
348                continue;
349            }
350
351            let result = self
352                .send_inform_to_sink(sink, trap_oid, uptime, &varbinds)
353                .await;
354            outcomes.push(SinkOutcome {
355                dest: sink.dest,
356                result,
357            });
358        }
359
360        NotificationOutcome { sinks: outcomes }
361    }
362
363    /// Send a trap PDU to a single sink.
364    async fn send_trap_to_sink(&self, sink: &TrapSink, pdu: &Pdu) -> Result<()> {
365        let data = match sink.version {
366            Version::V1 => {
367                // Convert the v2 PDU to a v1 TrapV1Pdu (RFC 3584 Section 3.2).
368                // Use the agent's bound address as agent_addr if available.
369                let local_ip = match self.inner.socket.local_addr() {
370                    Ok(addr) => match addr.ip() {
371                        std::net::IpAddr::V4(v4) => v4.octets(),
372                        std::net::IpAddr::V6(_) => [0, 0, 0, 0],
373                    },
374                    Err(_) => [0, 0, 0, 0],
375                };
376                let trap = pdu.to_v1_trap(local_ip).ok_or_else(|| {
377                    Error::Config("cannot convert trap to v1 for sink (Counter64 varbind?)".into())
378                        .boxed()
379                })?;
380                let msg = CommunityMessage::v1_trap(sink.community.clone(), trap);
381                msg.encode()
382            }
383            Version::V2c => {
384                let msg = CommunityMessage::new(Version::V2c, sink.community.clone(), pdu.clone());
385                msg.encode()
386            }
387            Version::V3 => {
388                let security = sink.v3_security.as_ref().ok_or_else(|| {
389                    Error::Config("V3 security not configured for trap sink".into()).boxed()
390                })?;
391
392                sink.ensure_keys_derived(&self.inner.state.engine_id)?;
393                let derived = sink.derived_keys.read().map_err(|_| {
394                    Error::Config("trap sink derived_keys lock poisoned".into()).boxed()
395                })?;
396
397                let (engine_boots, engine_time) = self.inner.state.authoritative_boots_time()?;
398
399                let msg_id = self.next_notification_id();
400                let encoded = crate::v3::encode::encode_v3_message(
401                    pdu,
402                    msg_id,
403                    &self.inner.state.engine_id,
404                    engine_boots,
405                    engine_time,
406                    security,
407                    derived.as_ref(),
408                    &self.inner.salt_counter,
409                    false, // reportable=false for traps
410                    crate::v3::DEFAULT_MSG_MAX_SIZE,
411                )?;
412                Bytes::from(encoded)
413            }
414        };
415
416        tracing::debug!(target: "async_snmp::agent", { snmp.dest = %sink.dest, snmp.bytes = data.len() }, "sending trap");
417        self.inner
418            .socket
419            .send_to(&data, sink.dest)
420            .await
421            .map_err(|e| Error::Network {
422                target: sink.dest,
423                source: e,
424            })?;
425
426        Ok(())
427    }
428
429    /// Send an inform to a single sink, reusing a cached client.
430    async fn send_inform_to_sink(
431        &self,
432        sink: &TrapSink,
433        trap_oid: &Oid,
434        uptime: u32,
435        varbinds: &[VarBind],
436    ) -> Result<()> {
437        let client = sink.get_or_create_inform_client().await?;
438        client
439            .send_inform(trap_oid, uptime, varbinds.to_vec())
440            .await?;
441
442        Ok(())
443    }
444
445    /// Generate a notification request/message ID.
446    fn next_notification_id(&self) -> i32 {
447        use std::sync::atomic::Ordering;
448        self.inner
449            .notification_id
450            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |v| {
451                Some(if v == i32::MAX { 1 } else { v + 1 })
452            })
453            .unwrap_or(1)
454    }
455}
456
457#[cfg(test)]
458mod tests {
459    use crate::agent::Agent;
460
461    #[tokio::test]
462    async fn test_notification_ids_are_per_agent() {
463        // Each Agent must own its own notification id sequence; two independent
464        // agents must not share a process-global counter.
465        let agent_a = Agent::builder()
466            .bind("127.0.0.1:0")
467            .community(b"public")
468            .build()
469            .await
470            .unwrap();
471        let agent_b = Agent::builder()
472            .bind("127.0.0.1:0")
473            .community(b"public")
474            .build()
475            .await
476            .unwrap();
477
478        // Advance agent_a's sequence a few times.
479        let a1 = agent_a.next_notification_id();
480        let a2 = agent_a.next_notification_id();
481        let a3 = agent_a.next_notification_id();
482        assert_eq!((a1, a2, a3), (1, 2, 3));
483
484        // agent_b is unaffected by agent_a's advancement and starts fresh.
485        let b1 = agent_b.next_notification_id();
486        let b2 = agent_b.next_notification_id();
487        assert_eq!((b1, b2), (1, 2));
488
489        // agent_a continues its own monotonic sequence.
490        assert_eq!(agent_a.next_notification_id(), 4);
491    }
492}
async_snmp/agent/notification.rs

async_snmp/agent/
notification.rs
