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

async_snmp/agent/
notification.rs
