async_snmp/client/
walk.rs

1//! Walk stream implementations.
2
3// Allow complex types for boxed futures in manual Stream implementations.
4// The `pending` fields require `Option<Pin<Box<dyn Future<Output = ...> + Send>>>`
5// which triggers this lint but is the standard pattern for storing futures.
6#![allow(clippy::type_complexity)]
7
8/// Implement `next()` and `collect()` for a Stream type that implements `poll_next`.
9macro_rules! impl_stream_helpers {
10    ($type:ident < $($gen:tt),+ >) => {
11        impl<$($gen),+> $type<$($gen),+>
12        where
13            $($gen: crate::transport::Transport + 'static,)+
14        {
15            /// Get the next varbind, or None when complete.
16            pub async fn next(&mut self) -> Option<crate::error::Result<crate::varbind::VarBind>> {
17                std::future::poll_fn(|cx| std::pin::Pin::new(&mut *self).poll_next(cx)).await
18            }
19
20            /// Collect all remaining varbinds.
21            pub async fn collect(mut self) -> crate::error::Result<Vec<crate::varbind::VarBind>> {
22                let mut results = Vec::new();
23                while let Some(result) = self.next().await {
24                    results.push(result?);
25                }
26                Ok(results)
27            }
28        }
29    };
30}
31
32use std::collections::{HashSet, VecDeque};
33use std::pin::Pin;
34use std::task::{Context, Poll};
35
36use futures_core::Stream;
37
38use crate::error::{Error, Result, WalkAbortReason};
39use crate::oid::Oid;
40use crate::transport::Transport;
41use crate::varbind::VarBind;
42use crate::version::Version;
43
44use super::Client;
45
46/// Walk operation mode.
47#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
48pub enum WalkMode {
49    /// Auto-select based on version (default).
50    /// V1 uses GETNEXT, V2c/V3 uses GETBULK.
51    #[default]
52    Auto,
53    /// Always use GETNEXT (slower but more compatible).
54    GetNext,
55    /// Always use GETBULK (faster, errors on v1).
56    GetBulk,
57}
58
59/// OID ordering behavior during walk operations.
60///
61/// SNMP walks rely on agents returning OIDs in strictly increasing
62/// lexicographic order. However, some buggy agents violate this requirement,
63/// returning OIDs out of order or even repeating OIDs (which would cause
64/// infinite loops).
65///
66/// This enum controls how the library handles ordering violations:
67///
68/// - [`Strict`](Self::Strict) (default): Terminates immediately with
69///   [`Error::WalkAborted`](crate::Error::WalkAborted) on any violation.
70///   Use this unless you know the agent has ordering bugs.
71///
72/// - [`AllowNonIncreasing`](Self::AllowNonIncreasing): Tolerates out-of-order
73///   OIDs but tracks all seen OIDs to detect cycles. Returns
74///   [`Error::WalkAborted`](crate::Error::WalkAborted) if the same OID appears twice.
75#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, PartialOrd, Ord, Hash)]
76pub enum OidOrdering {
77    /// Require strictly increasing OIDs (default).
78    ///
79    /// Walk terminates with [`Error::WalkAborted`](crate::Error::WalkAborted)
80    /// on first violation. Most efficient: O(1) memory, O(1) per-item check.
81    #[default]
82    Strict,
83
84    /// Allow non-increasing OIDs, with cycle detection.
85    ///
86    /// Some buggy agents return OIDs out of order. This mode tracks all seen
87    /// OIDs in a `HashSet` to detect cycles, terminating with an error if the
88    /// same OID is returned twice.
89    ///
90    /// **Warning**: This uses O(n) memory where n = number of walk results.
91    /// Always pair with [`ClientBuilder::max_walk_results`] to bound memory
92    /// usage. Cycle detection only catches duplicate OIDs; a pathological
93    /// agent could still return an infinite sequence of unique OIDs within
94    /// the subtree.
95    ///
96    /// [`ClientBuilder::max_walk_results`]: crate::ClientBuilder::max_walk_results
97    AllowNonIncreasing,
98}
99
100enum OidTracker {
101    Strict { last: Option<Oid> },
102    Relaxed { seen: HashSet<Oid> },
103}
104
105/// Outcome of validating a single varbind from a walk response.
106enum VarbindOutcome {
107    /// Varbind is valid and within the subtree; emit it.
108    Yield,
109    /// Walk is complete (end-of-MIB or out-of-subtree).
110    Done,
111    /// Walk should abort with the given error.
112    Abort(Box<Error>),
113}
114
115/// Validate a varbind received during a walk.
116///
117/// Checks end-of-MIB, subtree containment, and OID ordering.
118/// Returns the outcome, updating `oid_tracker` on success.
119fn validate_walk_varbind(
120    vb: &VarBind,
121    base_oid: &Oid,
122    oid_tracker: &mut OidTracker,
123    target: std::net::SocketAddr,
124) -> VarbindOutcome {
125    if vb.value.is_exception() {
126        return VarbindOutcome::Done;
127    }
128    if !vb.oid.starts_with(base_oid) {
129        return VarbindOutcome::Done;
130    }
131    match oid_tracker.check(&vb.oid, target) {
132        Ok(()) => VarbindOutcome::Yield,
133        Err(e) => VarbindOutcome::Abort(e),
134    }
135}
136
137impl OidTracker {
138    fn new(ordering: OidOrdering) -> Self {
139        match ordering {
140            OidOrdering::Strict => OidTracker::Strict { last: None },
141            OidOrdering::AllowNonIncreasing => OidTracker::Relaxed {
142                seen: HashSet::new(),
143            },
144        }
145    }
146
147    fn check(&mut self, oid: &Oid, target: std::net::SocketAddr) -> Result<()> {
148        match self {
149            OidTracker::Strict { last } => {
150                if let Some(prev) = last
151                    && oid <= prev
152                {
153                    tracing::debug!(target: "async_snmp::walk", { previous_oid = %prev, current_oid = %oid, %target }, "non-increasing OID detected");
154                    return Err(Error::WalkAborted {
155                        target,
156                        reason: WalkAbortReason::NonIncreasing,
157                    }
158                    .boxed());
159                }
160                *last = Some(oid.clone());
161                Ok(())
162            }
163            OidTracker::Relaxed { seen } => {
164                if !seen.insert(oid.clone()) {
165                    tracing::debug!(target: "async_snmp::walk", { %oid, %target }, "duplicate OID detected (cycle)");
166                    return Err(Error::WalkAborted {
167                        target,
168                        reason: WalkAbortReason::Cycle,
169                    }
170                    .boxed());
171                }
172                Ok(())
173            }
174        }
175    }
176}
177
178/// Async stream for walking an OID subtree using GETNEXT.
179///
180/// Created by [`Client::walk_getnext()`].
181pub struct Walk<T: Transport> {
182    client: Client<T>,
183    base_oid: Oid,
184    current_oid: Oid,
185    /// OID tracker for ordering validation.
186    oid_tracker: OidTracker,
187    /// Maximum number of results to return (None = unlimited).
188    max_results: Option<usize>,
189    /// Count of results returned so far.
190    count: usize,
191    done: bool,
192    pending: Option<Pin<Box<dyn std::future::Future<Output = Result<VarBind>> + Send>>>,
193}
194
195impl<T: Transport> Walk<T> {
196    pub(crate) fn new(
197        client: Client<T>,
198        oid: Oid,
199        ordering: OidOrdering,
200        max_results: Option<usize>,
201    ) -> Self {
202        Self {
203            client,
204            base_oid: oid.clone(),
205            current_oid: oid,
206            oid_tracker: OidTracker::new(ordering),
207            max_results,
208            count: 0,
209            done: false,
210            pending: None,
211        }
212    }
213}
214
215impl_stream_helpers!(Walk<T>);
216
217impl<T: Transport + 'static> Stream for Walk<T> {
218    type Item = Result<VarBind>;
219
220    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
221        if self.done {
222            return Poll::Ready(None);
223        }
224
225        // Check max_results limit
226        if let Some(max) = self.max_results
227            && self.count >= max
228        {
229            self.done = true;
230            return Poll::Ready(None);
231        }
232
233        // Check if we have a pending request
234        if self.pending.is_none() {
235            // Start a new GETNEXT request
236            let client = self.client.clone();
237            let oid = self.current_oid.clone();
238
239            let fut = Box::pin(async move { client.get_next(&oid).await });
240            self.pending = Some(fut);
241        }
242
243        // Poll the pending future
244        let pending = self.pending.as_mut().unwrap();
245        match pending.as_mut().poll(cx) {
246            Poll::Pending => Poll::Pending,
247            Poll::Ready(result) => {
248                self.pending = None;
249
250                match result {
251                    Ok(vb) => {
252                        let target = self.client.peer_addr();
253                        let base_oid = self.base_oid.clone();
254                        match validate_walk_varbind(&vb, &base_oid, &mut self.oid_tracker, target) {
255                            VarbindOutcome::Done => {
256                                self.done = true;
257                                return Poll::Ready(None);
258                            }
259                            VarbindOutcome::Abort(e) => {
260                                self.done = true;
261                                return Poll::Ready(Some(Err(e)));
262                            }
263                            VarbindOutcome::Yield => {}
264                        }
265
266                        // Update current OID for next iteration
267                        self.current_oid = vb.oid.clone();
268                        self.count += 1;
269
270                        Poll::Ready(Some(Ok(vb)))
271                    }
272                    Err(e) => {
273                        if self.client.inner.config.version == Version::V1
274                            && matches!(
275                                &*e,
276                                Error::Snmp {
277                                    status: crate::error::ErrorStatus::NoSuchName,
278                                    ..
279                                }
280                            )
281                        {
282                            self.done = true;
283                            return Poll::Ready(None);
284                        }
285
286                        self.done = true;
287                        Poll::Ready(Some(Err(e)))
288                    }
289                }
290            }
291        }
292    }
293}
294
295/// Async stream for walking an OID subtree using GETBULK.
296///
297/// Created by [`Client::bulk_walk()`].
298pub struct BulkWalk<T: Transport> {
299    client: Client<T>,
300    base_oid: Oid,
301    current_oid: Oid,
302    max_repetitions: i32,
303    /// OID tracker for ordering validation.
304    oid_tracker: OidTracker,
305    /// Maximum number of results to return (None = unlimited).
306    max_results: Option<usize>,
307    /// Count of results returned so far.
308    count: usize,
309    done: bool,
310    /// Buffered results from the last GETBULK response
311    buffer: VecDeque<VarBind>,
312    pending: Option<Pin<Box<dyn std::future::Future<Output = Result<Vec<VarBind>>> + Send>>>,
313}
314
315impl<T: Transport> BulkWalk<T> {
316    pub(crate) fn new(
317        client: Client<T>,
318        oid: Oid,
319        max_repetitions: i32,
320        ordering: OidOrdering,
321        max_results: Option<usize>,
322    ) -> Self {
323        Self {
324            client,
325            base_oid: oid.clone(),
326            current_oid: oid,
327            max_repetitions,
328            oid_tracker: OidTracker::new(ordering),
329            max_results,
330            count: 0,
331            done: false,
332            buffer: VecDeque::new(),
333            pending: None,
334        }
335    }
336}
337
338impl_stream_helpers!(BulkWalk<T>);
339
340impl<T: Transport + 'static> Stream for BulkWalk<T> {
341    type Item = Result<VarBind>;
342
343    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
344        loop {
345            if self.done {
346                return Poll::Ready(None);
347            }
348
349            // Check max_results limit
350            if let Some(max) = self.max_results
351                && self.count >= max
352            {
353                self.done = true;
354                return Poll::Ready(None);
355            }
356
357            // Check if we have buffered results to return
358            if let Some(vb) = self.buffer.pop_front() {
359                let target = self.client.peer_addr();
360                let base_oid = self.base_oid.clone();
361                match validate_walk_varbind(&vb, &base_oid, &mut self.oid_tracker, target) {
362                    VarbindOutcome::Done => {
363                        self.done = true;
364                        return Poll::Ready(None);
365                    }
366                    VarbindOutcome::Abort(e) => {
367                        self.done = true;
368                        return Poll::Ready(Some(Err(e)));
369                    }
370                    VarbindOutcome::Yield => {}
371                }
372
373                // Update current OID for next request
374                self.current_oid = vb.oid.clone();
375                self.count += 1;
376
377                return Poll::Ready(Some(Ok(vb)));
378            }
379
380            // Buffer exhausted, need to fetch more
381            if self.pending.is_none() {
382                let client = self.client.clone();
383                let oid = self.current_oid.clone();
384                let max_rep = self.max_repetitions;
385
386                let fut = Box::pin(async move { client.get_bulk(&[oid], 0, max_rep).await });
387                self.pending = Some(fut);
388            }
389
390            // Poll the pending future
391            let pending = self.pending.as_mut().unwrap();
392            match pending.as_mut().poll(cx) {
393                Poll::Pending => return Poll::Pending,
394                Poll::Ready(result) => {
395                    self.pending = None;
396
397                    match result {
398                        Ok(varbinds) => {
399                            if varbinds.is_empty() {
400                                self.done = true;
401                                return Poll::Ready(None);
402                            }
403
404                            self.buffer = varbinds.into();
405                            // Continue loop to process buffer
406                        }
407                        Err(e) => {
408                            self.done = true;
409                            return Poll::Ready(Some(Err(e)));
410                        }
411                    }
412                }
413            }
414        }
415    }
416}
417
418// ============================================================================
419// Unified WalkStream - auto-selects GETNEXT or GETBULK based on WalkMode
420// ============================================================================
421
422/// Unified walk stream that auto-selects between GETNEXT and GETBULK.
423///
424/// Created by [`Client::walk()`] when using `WalkMode::Auto` or explicit mode selection.
425/// This type wraps either a [`Walk`] or [`BulkWalk`] internally based on:
426/// - `WalkMode::Auto`: Uses GETNEXT for V1, GETBULK for V2c/V3
427/// - `WalkMode::GetNext`: Always uses GETNEXT
428/// - `WalkMode::GetBulk`: Always uses GETBULK (fails on V1)
429pub enum WalkStream<T: Transport> {
430    /// GETNEXT-based walk (used for V1 or when explicitly requested)
431    GetNext(Walk<T>),
432    /// GETBULK-based walk (used for V2c/V3 or when explicitly requested)
433    GetBulk(BulkWalk<T>),
434}
435
436impl<T: Transport> WalkStream<T> {
437    /// Create a new walk stream with auto-selection based on version and walk mode.
438    pub(crate) fn new(
439        client: Client<T>,
440        oid: Oid,
441        version: Version,
442        walk_mode: WalkMode,
443        ordering: OidOrdering,
444        max_results: Option<usize>,
445        max_repetitions: i32,
446    ) -> Result<Self> {
447        let use_bulk = match walk_mode {
448            WalkMode::Auto => version != Version::V1,
449            WalkMode::GetNext => false,
450            WalkMode::GetBulk => {
451                if version == Version::V1 {
452                    return Err(Error::Config("GETBULK is not supported in SNMPv1".into()).boxed());
453                }
454                true
455            }
456        };
457
458        Ok(if use_bulk {
459            WalkStream::GetBulk(BulkWalk::new(
460                client,
461                oid,
462                max_repetitions,
463                ordering,
464                max_results,
465            ))
466        } else {
467            WalkStream::GetNext(Walk::new(client, oid, ordering, max_results))
468        })
469    }
470}
471
472impl<T: Transport + 'static> WalkStream<T> {
473    /// Get the next varbind, or None when complete.
474    pub async fn next(&mut self) -> Option<Result<VarBind>> {
475        std::future::poll_fn(|cx| Pin::new(&mut *self).poll_next(cx)).await
476    }
477
478    /// Collect all remaining varbinds.
479    ///
480    /// If the walk completes with no results, a fallback GET is attempted on the
481    /// base OID. This handles scalar OIDs (e.g. `sysDescr.0`) where GETNEXT would
482    /// walk past the value. The GET result is only returned if it contains a real
483    /// value (not `NoSuchObject`, `NoSuchInstance`, or `EndOfMibView`).
484    pub async fn collect(mut self) -> Result<Vec<VarBind>> {
485        let mut results = Vec::new();
486        while let Some(result) = self.next().await {
487            results.push(result?);
488        }
489        if results.is_empty() {
490            let (client, base_oid) = match &self {
491                WalkStream::GetNext(w) => (&w.client, &w.base_oid),
492                WalkStream::GetBulk(bw) => (&bw.client, &bw.base_oid),
493            };
494            match client.get(base_oid).await {
495                Ok(vb) if !vb.value.is_exception() => {
496                    results.push(vb);
497                }
498                _ => {}
499            }
500        }
501        Ok(results)
502    }
503}
504
505impl<T: Transport + 'static> Stream for WalkStream<T> {
506    type Item = Result<VarBind>;
507
508    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
509        // SAFETY: We're just projecting the pin to the inner enum variant
510        match self.get_mut() {
511            WalkStream::GetNext(walk) => Pin::new(walk).poll_next(cx),
512            WalkStream::GetBulk(bulk_walk) => Pin::new(bulk_walk).poll_next(cx),
513        }
514    }
515}
516
517#[cfg(test)]
518mod tests {
519    use super::*;
520    use crate::oid;
521    use crate::value::Value;
522
523    fn target_addr() -> std::net::SocketAddr {
524        "127.0.0.1:161".parse().unwrap()
525    }
526
527    #[test]
528    fn test_walk_terminates_on_no_such_object() {
529        let base = oid!(1, 3, 6, 1, 2, 1, 1);
530        let mut tracker = OidTracker::new(OidOrdering::Strict);
531        let vb = VarBind::new(oid!(1, 3, 6, 1, 2, 1, 1, 1, 0), Value::NoSuchObject);
532        assert!(matches!(
533            validate_walk_varbind(&vb, &base, &mut tracker, target_addr()),
534            VarbindOutcome::Done
535        ));
536    }
537
538    #[test]
539    fn test_walk_terminates_on_no_such_instance() {
540        let base = oid!(1, 3, 6, 1, 2, 1, 1);
541        let mut tracker = OidTracker::new(OidOrdering::Strict);
542        let vb = VarBind::new(oid!(1, 3, 6, 1, 2, 1, 1, 1, 0), Value::NoSuchInstance);
543        assert!(matches!(
544            validate_walk_varbind(&vb, &base, &mut tracker, target_addr()),
545            VarbindOutcome::Done
546        ));
547    }
548
549    #[test]
550    fn test_walk_terminates_on_end_of_mib_view() {
551        let base = oid!(1, 3, 6, 1, 2, 1, 1);
552        let mut tracker = OidTracker::new(OidOrdering::Strict);
553        let vb = VarBind::new(oid!(1, 3, 6, 1, 2, 1, 1, 1, 0), Value::EndOfMibView);
554        assert!(matches!(
555            validate_walk_varbind(&vb, &base, &mut tracker, target_addr()),
556            VarbindOutcome::Done
557        ));
558    }
559
560    #[test]
561    fn test_walk_yields_normal_value() {
562        let base = oid!(1, 3, 6, 1, 2, 1, 1);
563        let mut tracker = OidTracker::new(OidOrdering::Strict);
564        let vb = VarBind::new(oid!(1, 3, 6, 1, 2, 1, 1, 1, 0), Value::Integer(42));
565        assert!(matches!(
566            validate_walk_varbind(&vb, &base, &mut tracker, target_addr()),
567            VarbindOutcome::Yield
568        ));
569    }
570
571    #[test]
572    fn test_walk_strict_aborts_on_non_increasing_oid() {
573        let base = oid!(1, 3, 6, 1, 2, 1, 1);
574        let mut tracker = OidTracker::new(OidOrdering::Strict);
575
576        let vb1 = VarBind::new(oid!(1, 3, 6, 1, 2, 1, 1, 2, 0), Value::Integer(1));
577        assert!(matches!(
578            validate_walk_varbind(&vb1, &base, &mut tracker, target_addr()),
579            VarbindOutcome::Yield
580        ));
581
582        // A lower in-subtree OID must abort with NonIncreasing.
583        let vb2 = VarBind::new(oid!(1, 3, 6, 1, 2, 1, 1, 1, 0), Value::Integer(2));
584        match validate_walk_varbind(&vb2, &base, &mut tracker, target_addr()) {
585            VarbindOutcome::Abort(e) => match *e {
586                Error::WalkAborted { reason, .. } => {
587                    assert_eq!(reason, WalkAbortReason::NonIncreasing);
588                }
589                other => panic!("expected WalkAborted, got {other:?}"),
590            },
591            _ => panic!("expected Abort outcome"),
592        }
593    }
594
595    #[test]
596    fn test_walk_strict_aborts_on_equal_oid() {
597        let base = oid!(1, 3, 6, 1, 2, 1, 1);
598        let mut tracker = OidTracker::new(OidOrdering::Strict);
599
600        let vb1 = VarBind::new(oid!(1, 3, 6, 1, 2, 1, 1, 2, 0), Value::Integer(1));
601        assert!(matches!(
602            validate_walk_varbind(&vb1, &base, &mut tracker, target_addr()),
603            VarbindOutcome::Yield
604        ));
605
606        // Same OID again (the `<=` boundary) must also abort with NonIncreasing.
607        let vb2 = VarBind::new(oid!(1, 3, 6, 1, 2, 1, 1, 2, 0), Value::Integer(1));
608        match validate_walk_varbind(&vb2, &base, &mut tracker, target_addr()) {
609            VarbindOutcome::Abort(e) => match *e {
610                Error::WalkAborted { reason, .. } => {
611                    assert_eq!(reason, WalkAbortReason::NonIncreasing);
612                }
613                other => panic!("expected WalkAborted, got {other:?}"),
614            },
615            _ => panic!("expected Abort outcome"),
616        }
617    }
618
619    #[test]
620    fn test_walk_relaxed_aborts_on_duplicate_oid_cycle() {
621        let base = oid!(1, 3, 6, 1, 2, 1, 1);
622        let mut tracker = OidTracker::new(OidOrdering::AllowNonIncreasing);
623
624        let vb1 = VarBind::new(oid!(1, 3, 6, 1, 2, 1, 1, 2, 0), Value::Integer(1));
625        assert!(matches!(
626            validate_walk_varbind(&vb1, &base, &mut tracker, target_addr()),
627            VarbindOutcome::Yield
628        ));
629
630        // Same OID again must abort with Cycle (not NonIncreasing).
631        let vb2 = VarBind::new(oid!(1, 3, 6, 1, 2, 1, 1, 2, 0), Value::Integer(1));
632        match validate_walk_varbind(&vb2, &base, &mut tracker, target_addr()) {
633            VarbindOutcome::Abort(e) => match *e {
634                Error::WalkAborted { reason, .. } => {
635                    assert_eq!(reason, WalkAbortReason::Cycle);
636                }
637                other => panic!("expected WalkAborted, got {other:?}"),
638            },
639            _ => panic!("expected Abort outcome"),
640        }
641    }
642
643    #[test]
644    fn test_walk_relaxed_allows_non_increasing_distinct_oid() {
645        let base = oid!(1, 3, 6, 1, 2, 1, 1);
646        let mut tracker = OidTracker::new(OidOrdering::AllowNonIncreasing);
647
648        // Higher OID first.
649        let vb1 = VarBind::new(oid!(1, 3, 6, 1, 2, 1, 1, 3, 0), Value::Integer(1));
650        assert!(matches!(
651            validate_walk_varbind(&vb1, &base, &mut tracker, target_addr()),
652            VarbindOutcome::Yield
653        ));
654
655        // Lower, but distinct, in-subtree OID: relaxed mode tolerates this (no abort).
656        let vb2 = VarBind::new(oid!(1, 3, 6, 1, 2, 1, 1, 1, 0), Value::Integer(2));
657        assert!(matches!(
658            validate_walk_varbind(&vb2, &base, &mut tracker, target_addr()),
659            VarbindOutcome::Yield
660        ));
661    }
662}
async_snmp/client/walk.rs

async_snmp/client/
walk.rs
