async_snmp/client/
mod.rs

1//! SNMP client implementation.
2
3mod auth;
4mod builder;
5mod retry;
6mod v3;
7mod walk;
8
9pub use auth::{Auth, CommunityVersion, UsmAuth, UsmBuilder};
10pub use builder::{ClientBuilder, Target};
11pub use retry::{Backoff, Retry, RetryBuilder};
12
13// New unified entry point
14impl Client<UdpHandle> {
15    /// Create a new SNMP client builder.
16    ///
17    /// This is the single entry point for client construction, supporting all
18    /// SNMP versions (v1, v2c, v3) through the [`Auth`] enum.
19    ///
20    /// # Example
21    ///
22    /// ```rust,no_run
23    /// use async_snmp::{Auth, Client, Retry};
24    /// use std::time::Duration;
25    ///
26    /// # async fn example() -> async_snmp::Result<()> {
27    /// // (host, port) tuple - convenient when host and port are separate
28    /// let client = Client::builder(("192.168.1.1", 161), Auth::v2c("public"))
29    ///     .connect().await?;
30    ///
31    /// // Combined address string (port defaults to 161 if omitted)
32    /// let client = Client::builder("switch.local", Auth::v2c("public"))
33    ///     .connect().await?;
34    ///
35    /// // SocketAddr works too
36    /// let addr: std::net::SocketAddr = "192.168.1.1:161".parse().unwrap();
37    /// let client = Client::builder(addr, Auth::v2c("public"))
38    ///     .connect().await?;
39    /// # Ok(())
40    /// # }
41    /// ```
42    pub fn builder(target: impl Into<Target>, auth: impl Into<Auth>) -> ClientBuilder {
43        ClientBuilder::new(target, auth)
44    }
45}
46use crate::error::internal::DecodeErrorKind;
47use crate::error::{Error, ErrorStatus, Result};
48use crate::message::{CommunityMessage, Message};
49use crate::oid::Oid;
50use crate::pdu::{GetBulkPdu, Pdu, PduType, TrapV1Pdu};
51use crate::transport::Transport;
52use crate::transport::UdpHandle;
53use crate::v3::{EngineCache, EngineState, SaltCounter};
54use crate::value::Value;
55use crate::varbind::VarBind;
56use crate::version::Version;
57use bytes::Bytes;
58use std::net::SocketAddr;
59use std::pin::Pin;
60use std::sync::Arc;
61use std::sync::RwLock;
62use std::time::{Duration, Instant};
63use tokio::sync::Mutex as AsyncMutex;
64use tracing::{Span, instrument};
65
66pub use crate::notification::{DerivedKeys, UsmConfig};
67pub use walk::{BulkWalk, OidOrdering, Walk, WalkMode, WalkStream};
68
69// ============================================================================
70// Shared helpers
71// ============================================================================
72
73/// Convert a configured `u32` `max_repetitions` to the `i32` GETBULK wire
74/// field, saturating instead of wrapping.
75///
76/// RFC 3416 Section 4.2.3 specifies `max-repetitions` as `INTEGER
77/// (0..2147483647)`. `ClientBuilder::max_repetitions` takes a `u32`, so a
78/// configured value above `i32::MAX` would wrap to a negative number under a
79/// plain `as i32` cast; this saturates to `i32::MAX` instead, which
80/// `GetBulkPdu::encode`'s own clamp (`.max(0)`) would otherwise silently turn
81/// into `0` (disabling repetitions).
82pub(crate) fn max_repetitions_to_wire(max_repetitions: u32) -> i32 {
83    i32::try_from(max_repetitions).unwrap_or(i32::MAX)
84}
85
86/// Extract an SNMP-level error from a PDU and convert it to an `Error::Snmp`.
87///
88/// Returns `Some(err)` if the PDU carries an SNMP error status, `None` otherwise.
89/// The `error_index` field is 1-based; 0 means the error applies to the whole PDU.
90pub(crate) fn pdu_to_snmp_error(pdu: &Pdu, target: SocketAddr) -> Option<Box<Error>> {
91    if !pdu.is_error() {
92        return None;
93    }
94    let status = pdu.error_status_enum();
95    let oid = (pdu.error_index as usize)
96        .checked_sub(1)
97        .and_then(|idx| pdu.varbinds.get(idx))
98        .map(|vb| vb.oid.clone());
99    Some(
100        Error::Snmp {
101            target,
102            status,
103            index: pdu.error_index.try_into().unwrap_or(0),
104            oid,
105        }
106        .boxed(),
107    )
108}
109
110// ============================================================================
111// Default configuration constants
112// ============================================================================
113
114/// Default timeout for SNMP requests.
115pub const DEFAULT_TIMEOUT: Duration = Duration::from_secs(5);
116
117/// Default maximum OIDs per request.
118///
119/// Requests with more OIDs than this limit are automatically split into
120/// multiple batches.
121pub const DEFAULT_MAX_OIDS_PER_REQUEST: usize = 10;
122
123/// Default max-repetitions for GETBULK operations.
124///
125/// Controls how many values are requested per GETBULK PDU during walks.
126pub const DEFAULT_MAX_REPETITIONS: u32 = 25;
127
128/// SNMP client.
129///
130/// Generic over transport type, with `UdpHandle` as default.
131pub struct Client<T: Transport = UdpHandle> {
132    inner: Arc<ClientInner<T>>,
133}
134
135impl<T: Transport> Clone for Client<T> {
136    fn clone(&self) -> Self {
137        Self {
138            inner: Arc::clone(&self.inner),
139        }
140    }
141}
142
143struct ClientInner<T: Transport> {
144    transport: T,
145    config: ClientConfig,
146    /// Cached engine state (V3)
147    engine_state: RwLock<Option<EngineState>>,
148    /// Derived keys for this engine (V3)
149    derived_keys: RwLock<Option<DerivedKeys>>,
150    /// Salt counter for privacy (V3)
151    salt_counter: SaltCounter,
152    /// Shared engine cache (V3, optional)
153    engine_cache: Option<Arc<EngineCache>>,
154    /// Serializes concurrent discovery attempts so only one runs at a time.
155    discovery_lock: AsyncMutex<()>,
156    /// Local engine start time for computing engine time in V3 traps.
157    local_engine_start: Instant,
158    /// Keys derived against `local_engine_id` for V3 trap sending.
159    local_derived_keys: RwLock<Option<DerivedKeys>>,
160}
161
162/// Client configuration.
163///
164/// Most users should use [`ClientBuilder`] rather than constructing this directly.
165#[derive(Clone)]
166pub struct ClientConfig {
167    /// SNMP version (default: V2c)
168    pub version: Version,
169    /// Community string for v1/v2c (default: "public")
170    pub community: Bytes,
171    /// Request timeout (default: 5 seconds)
172    pub timeout: Duration,
173    /// Retry configuration (default: 3 retries, 1-second delay)
174    pub retry: Retry,
175    /// Maximum OIDs per request (default: 10)
176    pub max_oids_per_request: usize,
177    /// `SNMPv3` security configuration (default: None)
178    pub v3_security: Option<UsmConfig>,
179    /// Walk operation mode (default: Auto)
180    pub walk_mode: WalkMode,
181    /// OID ordering behavior during walk operations (default: Strict)
182    pub oid_ordering: OidOrdering,
183    /// Maximum results from a single walk operation (default: None/unlimited)
184    pub max_walk_results: Option<usize>,
185    /// Max-repetitions for GETBULK operations (default: 25)
186    pub max_repetitions: u32,
187    /// Local engine ID for V3 trap sending (default: None).
188    ///
189    /// Per RFC 3412 Section 6.4, the sender is the authoritative engine for
190    /// trap PDUs. This engine ID is used to localize keys for outbound V3 traps.
191    pub local_engine_id: Option<Bytes>,
192    /// Local engine boots base value for V3 trap sending (default: 1).
193    pub local_engine_boots: u32,
194}
195
196impl Default for ClientConfig {
197    /// Returns configuration for `SNMPv2c` with community "public".
198    ///
199    /// See field documentation for all default values.
200    fn default() -> Self {
201        Self {
202            version: Version::V2c,
203            community: Bytes::from_static(b"public"),
204            timeout: DEFAULT_TIMEOUT,
205            retry: Retry::default(),
206            max_oids_per_request: DEFAULT_MAX_OIDS_PER_REQUEST,
207            v3_security: None,
208            walk_mode: WalkMode::Auto,
209            oid_ordering: OidOrdering::Strict,
210            max_walk_results: None,
211            max_repetitions: DEFAULT_MAX_REPETITIONS,
212            local_engine_id: None,
213            local_engine_boots: 1,
214        }
215    }
216}
217
218impl<T: Transport> Client<T> {
219    /// Create a new client with the given transport and config.
220    ///
221    /// For most use cases, prefer [`Client::builder()`] which provides a more
222    /// ergonomic API. Use this constructor when you need fine-grained control
223    /// over transport configuration (e.g., TCP connection timeout, keepalive
224    /// settings) or when using a custom [`Transport`] implementation.
225    pub fn new(transport: T, config: ClientConfig) -> Self {
226        Self {
227            inner: Arc::new(ClientInner {
228                transport,
229                config,
230                engine_state: RwLock::new(None),
231                derived_keys: RwLock::new(None),
232                salt_counter: SaltCounter::new(),
233                engine_cache: None,
234                discovery_lock: AsyncMutex::new(()),
235                local_engine_start: Instant::now(),
236                local_derived_keys: RwLock::new(None),
237            }),
238        }
239    }
240
241    /// Create a new V3 client with a shared engine cache.
242    pub fn with_engine_cache(
243        transport: T,
244        config: ClientConfig,
245        engine_cache: Arc<EngineCache>,
246    ) -> Self {
247        Self {
248            inner: Arc::new(ClientInner {
249                transport,
250                config,
251                engine_state: RwLock::new(None),
252                derived_keys: RwLock::new(None),
253                salt_counter: SaltCounter::new(),
254                engine_cache: Some(engine_cache),
255                discovery_lock: AsyncMutex::new(()),
256                local_engine_start: Instant::now(),
257                local_derived_keys: RwLock::new(None),
258            }),
259        }
260    }
261
262    /// Get the peer (target) address.
263    ///
264    /// Returns the remote address that this client sends requests to.
265    /// Named to match [`std::net::TcpStream::peer_addr()`].
266    #[must_use]
267    pub fn peer_addr(&self) -> SocketAddr {
268        self.inner.transport.peer_addr()
269    }
270
271    /// Generate next request ID.
272    ///
273    /// Uses the transport's allocator (backed by a global counter).
274    fn next_request_id(&self) -> i32 {
275        self.inner.transport.alloc_request_id()
276    }
277
278    /// Check if using V3 with authentication/encryption configured.
279    fn is_v3(&self) -> bool {
280        self.inner.config.version == Version::V3 && self.inner.config.v3_security.is_some()
281    }
282
283    /// Send a request and wait for response (internal helper with pre-encoded data).
284    #[instrument(
285        level = "debug",
286        skip(self, data),
287        fields(
288            snmp.target = %self.peer_addr(),
289            snmp.request_id = request_id,
290            snmp.attempt = tracing::field::Empty,
291            snmp.elapsed_ms = tracing::field::Empty,
292        )
293    )]
294    async fn send_and_recv(&self, request_id: i32, data: &[u8]) -> Result<Pdu> {
295        let start = Instant::now();
296        let mut last_error: Option<Box<Error>> = None;
297        let max_attempts = if self.inner.transport.is_reliable() {
298            0
299        } else {
300            self.inner.config.retry.max_attempts
301        };
302
303        for attempt in 0..=max_attempts {
304            Span::current().record("snmp.attempt", attempt);
305            if attempt > 0 {
306                tracing::debug!(target: "async_snmp::client", "retrying request");
307            }
308
309            // Register (or re-register) with fresh deadline before sending
310            self.inner
311                .transport
312                .register_request(request_id, self.inner.config.timeout);
313
314            // Send request
315            tracing::trace!(target: "async_snmp::client", { snmp.bytes = data.len() }, "sending request");
316            self.inner.transport.send(data).await?;
317
318            // Wait for response (deadline was set by register_request)
319            match self.inner.transport.recv(request_id).await {
320                Ok((response_data, _source)) => {
321                    tracing::trace!(target: "async_snmp::client", { snmp.bytes = response_data.len() }, "received response");
322
323                    // Decode response and extract PDU
324                    let response = Message::decode(response_data)?;
325
326                    // Validate response version matches request version
327                    let response_version = response.version();
328                    let expected_version = self.inner.config.version;
329                    if response_version != expected_version {
330                        tracing::warn!(target: "async_snmp::client", { ?expected_version, ?response_version, peer = %self.peer_addr() }, "version mismatch in response");
331                        return Err(Error::MalformedResponse {
332                            target: self.peer_addr(),
333                        }
334                        .boxed());
335                    }
336
337                    // Warn when the community does not echo the one we sent.
338                    // net-snmp accepts such responses (proxies and some
339                    // agents rewrite the community), so this is not a reject.
340                    if let Message::Community(ref m) = response
341                        && m.community != self.inner.config.community
342                    {
343                        tracing::warn!(target: "async_snmp::client", { peer = %self.peer_addr() }, "community mismatch in response");
344                    }
345
346                    let Some(response_pdu) = response.into_pdu() else {
347                        tracing::warn!(target: "async_snmp::client", { peer = %self.peer_addr() }, "received TrapV1 in response to request");
348                        return Err(Error::MalformedResponse {
349                            target: self.peer_addr(),
350                        }
351                        .boxed());
352                    };
353
354                    // RFC 3416 Section 4.2: only a Response-PDU may answer a
355                    // request; reject echoed request-type PDUs
356                    if response_pdu.pdu_type != PduType::Response {
357                        tracing::warn!(target: "async_snmp::client", { peer = %self.peer_addr(), pdu_type = ?response_pdu.pdu_type }, "non-Response PDU in response");
358                        return Err(Error::MalformedResponse {
359                            target: self.peer_addr(),
360                        }
361                        .boxed());
362                    }
363
364                    // Validate request ID
365                    if response_pdu.request_id != request_id {
366                        tracing::warn!(target: "async_snmp::client", { expected_request_id = request_id, actual_request_id = response_pdu.request_id, peer = %self.peer_addr() }, "request ID mismatch in response");
367                        return Err(Error::MalformedResponse {
368                            target: self.peer_addr(),
369                        }
370                        .boxed());
371                    }
372
373                    // Check for SNMP error
374                    if let Some(err) = pdu_to_snmp_error(&response_pdu, self.peer_addr()) {
375                        Span::current()
376                            .record("snmp.elapsed_ms", start.elapsed().as_millis() as u64);
377                        return Err(err);
378                    }
379
380                    Span::current().record("snmp.elapsed_ms", start.elapsed().as_millis() as u64);
381                    return Ok(response_pdu);
382                }
383                Err(e) if matches!(*e, Error::Timeout { .. }) => {
384                    last_error = Some(e);
385                    // Apply backoff delay before next retry (if not last attempt)
386                    if attempt < max_attempts {
387                        let delay = self.inner.config.retry.compute_delay(attempt);
388                        if !delay.is_zero() {
389                            tracing::debug!(target: "async_snmp::client", { delay_ms = delay.as_millis() as u64 }, "backing off");
390                            tokio::time::sleep(delay).await;
391                        }
392                    }
393                    // fall thru to next loop iteration
394                }
395                Err(e) => {
396                    Span::current().record("snmp.elapsed_ms", start.elapsed().as_millis() as u64);
397                    return Err(e);
398                }
399            }
400        }
401
402        // All retries exhausted
403        let elapsed = start.elapsed();
404        Span::current().record("snmp.elapsed_ms", elapsed.as_millis() as u64);
405        tracing::debug!(target: "async_snmp::client", { request_id, peer = %self.peer_addr(), ?elapsed, retries = max_attempts }, "request timed out");
406        Err(last_error.unwrap_or_else(|| {
407            Error::Timeout {
408                target: self.peer_addr(),
409                elapsed,
410                retries: max_attempts,
411            }
412            .boxed()
413        }))
414    }
415
416    /// Send a standard request (GET, GETNEXT, SET) and wait for response.
417    async fn send_request(&self, pdu: Pdu) -> Result<Pdu> {
418        // Dispatch to V3 handler if configured
419        if self.is_v3() {
420            return self.send_v3_and_recv(pdu).await;
421        }
422
423        tracing::debug!(target: "async_snmp::client", { snmp.pdu_type = ?pdu.pdu_type, snmp.varbind_count = pdu.varbinds.len() }, "sending {} request", pdu.pdu_type);
424
425        let request_id = pdu.request_id;
426        let message = CommunityMessage::new(
427            self.inner.config.version,
428            self.inner.config.community.clone(),
429            pdu,
430        );
431        let data = message.encode();
432        let response = self.send_and_recv(request_id, &data).await?;
433
434        tracing::debug!(target: "async_snmp::client", { snmp.pdu_type = ?response.pdu_type, snmp.varbind_count = response.varbinds.len(), snmp.error_status = response.error_status, snmp.error_index = response.error_index }, "received {} response", response.pdu_type);
435
436        Ok(response)
437    }
438
439    /// Send a GETBULK request and wait for response.
440    async fn send_bulk_request(&self, pdu: GetBulkPdu) -> Result<Pdu> {
441        // Dispatch to V3 handler if configured
442        if self.is_v3() {
443            // Convert GetBulkPdu to Pdu for V3 encoding
444            let pdu = Pdu::get_bulk(
445                pdu.request_id,
446                pdu.non_repeaters,
447                pdu.max_repetitions,
448                pdu.varbinds,
449            );
450            return self.send_v3_and_recv(pdu).await;
451        }
452
453        tracing::debug!(target: "async_snmp::client", { snmp.non_repeaters = pdu.non_repeaters, snmp.max_repetitions = pdu.max_repetitions, snmp.varbind_count = pdu.varbinds.len() }, "sending GetBulkRequest");
454
455        let request_id = pdu.request_id;
456        let data = CommunityMessage::encode_bulk(
457            self.inner.config.version,
458            self.inner.config.community.clone(),
459            &pdu,
460        );
461        let response = self.send_and_recv(request_id, &data).await?;
462
463        tracing::debug!(target: "async_snmp::client", { snmp.pdu_type = ?response.pdu_type, snmp.varbind_count = response.varbinds.len(), snmp.error_status = response.error_status, snmp.error_index = response.error_index }, "received {} response", response.pdu_type);
464
465        Ok(response)
466    }
467
468    /// GET a single OID.
469    #[instrument(skip(self), err, fields(snmp.target = %self.peer_addr(), snmp.oid = %oid))]
470    pub async fn get(&self, oid: &Oid) -> Result<VarBind> {
471        let request_id = self.next_request_id();
472        let pdu = Pdu::get_request(request_id, std::slice::from_ref(oid));
473        let response = self.send_request(pdu).await?;
474
475        response.varbinds.into_iter().next().ok_or_else(|| {
476            tracing::debug!(target: "async_snmp::client", { peer = %self.peer_addr(), kind = %DecodeErrorKind::EmptyResponse }, "empty GET response");
477            Error::MalformedResponse {
478                target: self.peer_addr(),
479            }
480            .boxed()
481        })
482    }
483
484    /// GET multiple OIDs.
485    ///
486    /// If the OID list exceeds `max_oids_per_request`, the request is
487    /// automatically split into multiple batches. Results are returned
488    /// in the same order as the input OIDs.
489    ///
490    /// # Example
491    ///
492    /// ```rust,no_run
493    /// # use async_snmp::{Auth, Client, oid};
494    /// # async fn example() -> async_snmp::Result<()> {
495    /// # let client = Client::builder("127.0.0.1:161", Auth::v2c("public")).connect().await?;
496    /// let results = client.get_many(&[
497    ///     oid!(1, 3, 6, 1, 2, 1, 1, 1, 0),  // sysDescr
498    ///     oid!(1, 3, 6, 1, 2, 1, 1, 3, 0),  // sysUpTime
499    ///     oid!(1, 3, 6, 1, 2, 1, 1, 5, 0),  // sysName
500    /// ]).await?;
501    /// # Ok(())
502    /// # }
503    /// ```
504    #[instrument(skip(self, oids), err, fields(snmp.target = %self.peer_addr(), snmp.oid_count = oids.len()))]
505    pub async fn get_many(&self, oids: &[Oid]) -> Result<Vec<VarBind>> {
506        self.get_or_getnext_many(oids, "GET", Pdu::get_request)
507            .await
508    }
509
510    /// GETNEXT for a single OID.
511    #[instrument(skip(self), err, fields(snmp.target = %self.peer_addr(), snmp.oid = %oid))]
512    pub async fn get_next(&self, oid: &Oid) -> Result<VarBind> {
513        let request_id = self.next_request_id();
514        let pdu = Pdu::get_next_request(request_id, std::slice::from_ref(oid));
515        let response = self.send_request(pdu).await?;
516
517        response.varbinds.into_iter().next().ok_or_else(|| {
518            tracing::debug!(target: "async_snmp::client", { peer = %self.peer_addr(), kind = %DecodeErrorKind::EmptyResponse }, "empty GETNEXT response");
519            Error::MalformedResponse {
520                target: self.peer_addr(),
521            }
522            .boxed()
523        })
524    }
525
526    /// GETNEXT for multiple OIDs.
527    ///
528    /// If the OID list exceeds `max_oids_per_request`, the request is
529    /// automatically split into multiple batches. Results are returned
530    /// in the same order as the input OIDs.
531    ///
532    /// # Example
533    ///
534    /// ```rust,no_run
535    /// # use async_snmp::{Auth, Client, oid};
536    /// # async fn example() -> async_snmp::Result<()> {
537    /// # let client = Client::builder("127.0.0.1:161", Auth::v2c("public")).connect().await?;
538    /// let results = client.get_next_many(&[
539    ///     oid!(1, 3, 6, 1, 2, 1, 2, 2, 1, 2),  // ifDescr
540    ///     oid!(1, 3, 6, 1, 2, 1, 2, 2, 1, 3),  // ifType
541    /// ]).await?;
542    /// # Ok(())
543    /// # }
544    /// ```
545    #[instrument(skip(self, oids), err, fields(snmp.target = %self.peer_addr(), snmp.oid_count = oids.len()))]
546    pub async fn get_next_many(&self, oids: &[Oid]) -> Result<Vec<VarBind>> {
547        self.get_or_getnext_many(oids, "GETNEXT", Pdu::get_next_request)
548            .await
549    }
550
551    /// Shared implementation for GET-many and GETNEXT-many.
552    ///
553    /// `op` is the PDU constructor (`Pdu::get_request` or `Pdu::get_next_request`).
554    /// `op_name` is used only for log messages.
555    async fn get_or_getnext_many(
556        &self,
557        oids: &[Oid],
558        op_name: &'static str,
559        op: fn(i32, &[Oid]) -> Pdu,
560    ) -> Result<Vec<VarBind>> {
561        if oids.is_empty() {
562            return Ok(Vec::new());
563        }
564
565        let max_per_request = self.inner.config.max_oids_per_request;
566        let mut all_results = Vec::with_capacity(oids.len());
567
568        for chunk in oids.chunks(max_per_request) {
569            self.send_batch_with_bisect(chunk, op_name, op, &mut all_results)
570                .await?;
571        }
572
573        Ok(all_results)
574    }
575
576    /// Send a batch of OIDs, automatically bisecting on tooBig errors.
577    ///
578    /// If the agent returns tooBig for a batch with more than one OID, the batch
579    /// is split in half and each half is retried. This repeats recursively until
580    /// batches succeed or a single-OID request fails (which is unrecoverable).
581    fn send_batch_with_bisect<'a>(
582        &'a self,
583        oids: &'a [Oid],
584        op_name: &'static str,
585        op: fn(i32, &[Oid]) -> Pdu,
586        results: &'a mut Vec<VarBind>,
587    ) -> Pin<Box<dyn std::future::Future<Output = Result<()>> + Send + 'a>> {
588        Box::pin(async move {
589            let request_id = self.next_request_id();
590            let pdu = op(request_id, oids);
591            match self.send_request(pdu).await {
592                Ok(response) => {
593                    if response.varbinds.len() > oids.len() {
594                        tracing::warn!(target: "async_snmp::client", { peer = %self.peer_addr(), expected = oids.len(), actual = response.varbinds.len(), snmp.op = op_name }, "response has more varbinds than requested");
595                        return Err(Error::MalformedResponse {
596                            target: self.peer_addr(),
597                        }
598                        .boxed());
599                    } else if response.varbinds.len() < oids.len() {
600                        tracing::warn!(target: "async_snmp::client", { peer = %self.peer_addr(), expected = oids.len(), actual = response.varbinds.len(), snmp.op = op_name }, "response has fewer varbinds than requested");
601                    }
602                    results.extend(response.varbinds);
603                    Ok(())
604                }
605                Err(e)
606                    if oids.len() > 1
607                        && matches!(
608                            &*e,
609                            Error::Snmp {
610                                status: ErrorStatus::TooBig,
611                                ..
612                            }
613                        ) =>
614                {
615                    let mid = oids.len() / 2;
616                    tracing::debug!(target: "async_snmp::client", { peer = %self.peer_addr(), snmp.batch_size = oids.len(), snmp.split_at = mid, snmp.op = op_name }, "tooBig response, bisecting batch");
617                    self.send_batch_with_bisect(&oids[..mid], op_name, op, results)
618                        .await?;
619                    self.send_batch_with_bisect(&oids[mid..], op_name, op, results)
620                        .await?;
621                    Ok(())
622                }
623                Err(e) => Err(e),
624            }
625        })
626    }
627
628    /// SET a single OID.
629    #[instrument(skip(self, value), err, fields(snmp.target = %self.peer_addr(), snmp.oid = %oid))]
630    pub async fn set(&self, oid: &Oid, value: Value) -> Result<VarBind> {
631        let request_id = self.next_request_id();
632        let varbind = VarBind::new(oid.clone(), value);
633        let pdu = Pdu::set_request(request_id, vec![varbind]);
634        let response = self.send_request(pdu).await?;
635
636        response.varbinds.into_iter().next().ok_or_else(|| {
637            tracing::debug!(target: "async_snmp::client", { peer = %self.peer_addr(), kind = %DecodeErrorKind::EmptyResponse }, "empty SET response");
638            Error::MalformedResponse {
639                target: self.peer_addr(),
640            }
641            .boxed()
642        })
643    }
644
645    /// SET multiple OIDs in a single atomic PDU.
646    ///
647    /// RFC 3416 requires that a SET request be atomic: either all variables
648    /// in the request are set, or none are. To preserve this guarantee,
649    /// `set_many` refuses to split the varbind list across multiple PDUs.
650    ///
651    /// If `varbinds.len()` exceeds `max_oids_per_request`, this method
652    /// returns `Error::Config` rather than silently batching the request.
653    /// Callers that need to set more variables than the per-request limit
654    /// must issue multiple explicit `set_many` calls and handle partial
655    /// failure themselves.
656    ///
657    /// # Example
658    ///
659    /// ```rust,no_run
660    /// # use async_snmp::{Auth, Client, oid, Value};
661    /// # async fn example() -> async_snmp::Result<()> {
662    /// # let client = Client::builder("127.0.0.1:161", Auth::v2c("private")).connect().await?;
663    /// let results = client.set_many(&[
664    ///     (oid!(1, 3, 6, 1, 2, 1, 1, 5, 0), Value::from("new-hostname")),
665    ///     (oid!(1, 3, 6, 1, 2, 1, 1, 6, 0), Value::from("new-location")),
666    /// ]).await?;
667    /// # Ok(())
668    /// # }
669    /// ```
670    #[instrument(skip(self, varbinds), err, fields(snmp.target = %self.peer_addr(), snmp.oid_count = varbinds.len()))]
671    pub async fn set_many(&self, varbinds: &[(Oid, Value)]) -> Result<Vec<VarBind>> {
672        if varbinds.is_empty() {
673            return Ok(Vec::new());
674        }
675
676        let max_per_request = self.inner.config.max_oids_per_request;
677
678        if varbinds.len() > max_per_request {
679            return Err(Error::Config(
680                format!(
681                    "set_many: {} varbinds exceeds max_oids_per_request ({}); \
682                     SET must be atomic and cannot be split across PDUs",
683                    varbinds.len(),
684                    max_per_request,
685                )
686                .into(),
687            )
688            .boxed());
689        }
690
691        let request_id = self.next_request_id();
692        let vbs: Vec<VarBind> = varbinds
693            .iter()
694            .map(|(oid, value)| VarBind::new(oid.clone(), value.clone()))
695            .collect();
696        let expected_count = vbs.len();
697        let pdu = Pdu::set_request(request_id, vbs);
698        let response = self.send_request(pdu).await?;
699        if response.varbinds.len() > expected_count {
700            tracing::warn!(target: "async_snmp::client", { peer = %self.peer_addr(), expected = expected_count, actual = response.varbinds.len() }, "SET response has more varbinds than requested");
701            return Err(Error::MalformedResponse {
702                target: self.peer_addr(),
703            }
704            .boxed());
705        } else if response.varbinds.len() < expected_count {
706            tracing::warn!(target: "async_snmp::client", { peer = %self.peer_addr(), expected = expected_count, actual = response.varbinds.len() }, "SET response has fewer varbinds than requested");
707        }
708        Ok(response.varbinds)
709    }
710
711    /// Send a trap (fire-and-forget).
712    ///
713    /// For V1 clients: constructs a `TrapV1` PDU. The `trap_oid` is reverse-mapped
714    /// to v1 `generic_trap/specific_trap/enterprise` fields per RFC 3584 Section 3.2.
715    /// The `agent_addr` is set from the transport's local IPv4 address, or `[0,0,0,0]`
716    /// if the local address is IPv6. Use [`send_v1_trap`](Self::send_v1_trap) for
717    /// explicit control over v1 fields.
718    ///
719    /// For V2c/V3 clients: constructs a `TrapV2` PDU with the mandatory sysUpTime.0
720    /// and snmpTrapOID.0 prefix.
721    ///
722    /// For V3: uses the local engine ID (set via `ClientBuilder::local_engine_id`).
723    ///
724    /// # Arguments
725    ///
726    /// * `trap_oid` - The trap OID (snmpTrapOID.0 value)
727    /// * `uptime` - sysUpTime.0 value in hundredths of seconds
728    /// * `varbinds` - Additional variable bindings (appended after the prefix)
729    #[instrument(skip(self, varbinds), err, fields(snmp.target = %self.peer_addr(), snmp.trap_oid = %trap_oid))]
730    pub async fn send_trap(
731        &self,
732        trap_oid: &Oid,
733        uptime: u32,
734        varbinds: Vec<VarBind>,
735    ) -> Result<()> {
736        if self.inner.config.version == Version::V1 {
737            // Build a v2-style PDU and convert to v1.
738            // Per RFC 3584 Section 3, use the local IPv4 address as agent_addr.
739            let local_ip = match self.inner.transport.local_addr().ip() {
740                std::net::IpAddr::V4(v4) => v4.octets(),
741                std::net::IpAddr::V6(_) => [0, 0, 0, 0],
742            };
743            // request_id is unused in the v1 wire format, use 0 to avoid
744            // wasting a slot in the request_id sequence.
745            let pdu = Pdu::trap_v2(0, uptime, trap_oid, varbinds);
746            let trap = pdu.to_v1_trap(local_ip).ok_or_else(|| {
747                Error::Config("cannot convert trap to v1 (Counter64 varbind?)".into()).boxed()
748            })?;
749            return self.send_v1_trap(trap).await;
750        }
751
752        let request_id = self.next_request_id();
753        let pdu = Pdu::trap_v2(request_id, uptime, trap_oid, varbinds);
754
755        if self.is_v3() {
756            self.ensure_local_keys_derived()?;
757            let msg_id = self.next_request_id();
758            let data = self.build_v3_trap_message(&pdu, msg_id)?;
759            tracing::debug!(target: "async_snmp::client", { snmp.pdu_type = "TrapV2", snmp.varbind_count = pdu.varbinds.len(), snmp.bytes = data.len() }, "sending V3 trap");
760            self.inner.transport.send(&data).await?;
761        } else {
762            let message = CommunityMessage::new(
763                self.inner.config.version,
764                self.inner.config.community.clone(),
765                pdu,
766            );
767            let data = message.encode();
768            tracing::debug!(target: "async_snmp::client", { snmp.pdu_type = "TrapV2", snmp.bytes = data.len() }, "sending v2c trap");
769            self.inner.transport.send(&data).await?;
770        }
771
772        Ok(())
773    }
774
775    /// Send an `SNMPv1` trap with explicit v1 PDU fields.
776    ///
777    /// This is a lower-level method that accepts a pre-built [`TrapV1Pdu`],
778    /// giving full control over enterprise OID, `agent_addr`, `generic_trap`,
779    /// `specific_trap`, and `time_stamp` fields.
780    ///
781    /// The client must be configured for V1 (`Auth::v1()`). Returns an error
782    /// if the client version is not V1.
783    ///
784    /// # Example
785    ///
786    /// ```rust,no_run
787    /// # use async_snmp::{Auth, Client, TrapV1Pdu, GenericTrap, oid};
788    /// # async fn example() -> async_snmp::Result<()> {
789    /// let client = Client::builder("192.168.1.100:162", Auth::v1("public"))
790    ///     .connect().await?;
791    ///
792    /// let trap = TrapV1Pdu::new(
793    ///     oid!(1, 3, 6, 1, 4, 1, 9999),  // enterprise
794    ///     [192, 168, 1, 1],               // agent address
795    ///     GenericTrap::ColdStart,
796    ///     0,
797    ///     12345,                          // uptime in centiseconds
798    ///     vec![],
799    /// );
800    /// client.send_v1_trap(trap).await?;
801    /// # Ok(())
802    /// # }
803    /// ```
804    #[instrument(skip(self, trap), err, fields(snmp.target = %self.peer_addr(), snmp.generic_trap = %trap.generic_trap))]
805    pub async fn send_v1_trap(&self, trap: TrapV1Pdu) -> Result<()> {
806        if self.inner.config.version != Version::V1 {
807            return Err(Error::Config("send_v1_trap requires a V1 client".into()).boxed());
808        }
809
810        let message = CommunityMessage::v1_trap(self.inner.config.community.clone(), trap);
811        let data = message.encode();
812        tracing::debug!(target: "async_snmp::client", { snmp.pdu_type = "TrapV1", snmp.bytes = data.len() }, "sending v1 trap");
813        self.inner.transport.send(&data).await?;
814
815        Ok(())
816    }
817
818    /// Send a v2c/v3 inform and wait for acknowledgement.
819    ///
820    /// Constructs an `InformRequest` PDU with the mandatory sysUpTime.0 and
821    /// snmpTrapOID.0 prefix, sends it to the target, and waits for a Response
822    /// PDU. Uses the same retry and timeout logic as other request types.
823    ///
824    /// For V3: uses engine discovery against the receiver (same as GET/SET).
825    /// V1 is not supported and returns an error.
826    ///
827    /// # Arguments
828    ///
829    /// * `trap_oid` - The trap OID (snmpTrapOID.0 value)
830    /// * `uptime` - sysUpTime.0 value in hundredths of seconds
831    /// * `varbinds` - Additional variable bindings (appended after the prefix)
832    #[instrument(skip(self, varbinds), err, fields(snmp.target = %self.peer_addr(), snmp.trap_oid = %trap_oid))]
833    pub async fn send_inform(
834        &self,
835        trap_oid: &Oid,
836        uptime: u32,
837        varbinds: Vec<VarBind>,
838    ) -> Result<()> {
839        if self.inner.config.version == Version::V1 {
840            return Err(Error::Config("v1 inform sending not supported".into()).boxed());
841        }
842
843        let request_id = self.next_request_id();
844        let pdu = Pdu::inform_request(request_id, uptime, trap_oid, varbinds);
845        let _response = self.send_request(pdu).await?;
846        Ok(())
847    }
848
849    /// GETBULK request (SNMPv2c/v3 only).
850    ///
851    /// Efficiently retrieves multiple variable bindings in a single request.
852    /// GETBULK splits the requested OIDs into two groups:
853    ///
854    /// - **Non-repeaters** (first N OIDs): Each gets a single GETNEXT, returning
855    ///   one value per OID. Use for scalar values like `sysUpTime.0`.
856    /// - **Repeaters** (remaining OIDs): Each gets up to `max_repetitions` GETNEXTs,
857    ///   returning multiple values per OID. Use for walking table columns.
858    ///
859    /// # Arguments
860    ///
861    /// * `oids` - OIDs to retrieve
862    /// * `non_repeaters` - How many OIDs (from the start) are non-repeating
863    /// * `max_repetitions` - Maximum rows to return for each repeating OID
864    ///
865    /// # Example
866    ///
867    /// ```rust,no_run
868    /// # use async_snmp::{Auth, Client, oid};
869    /// # async fn example() -> async_snmp::Result<()> {
870    /// # let client = Client::builder("127.0.0.1:161", Auth::v2c("public")).connect().await?;
871    /// // Get sysUpTime (non-repeater) plus 10 interface descriptions (repeater)
872    /// let results = client.get_bulk(
873    ///     &[oid!(1, 3, 6, 1, 2, 1, 1, 3, 0), oid!(1, 3, 6, 1, 2, 1, 2, 2, 1, 2)],
874    ///     1,  // first OID is non-repeating
875    ///     10, // get up to 10 values for the second OID
876    /// ).await?;
877    /// // Results: [sysUpTime value, ifDescr.1, ifDescr.2, ..., ifDescr.10]
878    /// # Ok(())
879    /// # }
880    /// ```
881    #[instrument(skip(self, oids), err, fields(
882        snmp.target = %self.peer_addr(),
883        snmp.oid_count = oids.len(),
884        snmp.non_repeaters = non_repeaters,
885        snmp.max_repetitions = max_repetitions
886    ))]
887    pub async fn get_bulk(
888        &self,
889        oids: &[Oid],
890        non_repeaters: i32,
891        max_repetitions: i32,
892    ) -> Result<Vec<VarBind>> {
893        let request_id = self.next_request_id();
894        let pdu = GetBulkPdu::new(request_id, non_repeaters, max_repetitions, oids);
895        let response = self.send_bulk_request(pdu).await?;
896        Ok(response.varbinds)
897    }
898
899    /// Walk an OID subtree.
900    ///
901    /// Auto-selects the optimal walk method based on SNMP version and `WalkMode`:
902    /// - `WalkMode::Auto` (default): Uses GETNEXT for V1, GETBULK for V2c/V3
903    /// - `WalkMode::GetNext`: Always uses GETNEXT
904    /// - `WalkMode::GetBulk`: Always uses GETBULK (fails on V1)
905    ///
906    /// Returns an async stream that yields each variable binding in the subtree.
907    /// The walk terminates when an OID outside the subtree is encountered or
908    /// when `EndOfMibView` is returned.
909    ///
910    /// Uses the client's configured `oid_ordering`, `max_walk_results`, and
911    /// `max_repetitions` (for GETBULK) settings.
912    ///
913    /// # Example
914    ///
915    /// ```rust,no_run
916    /// # use async_snmp::{Auth, Client, oid};
917    /// # async fn example() -> async_snmp::Result<()> {
918    /// # let client = Client::builder("127.0.0.1:161", Auth::v2c("public")).connect().await?;
919    /// // Auto-selects GETBULK for V2c/V3, GETNEXT for V1
920    /// let results = client.walk(oid!(1, 3, 6, 1, 2, 1, 1))?.collect().await?;
921    /// # Ok(())
922    /// # }
923    /// ```
924    #[instrument(skip(self), fields(snmp.target = %self.peer_addr(), snmp.oid = %oid))]
925    pub fn walk(&self, oid: Oid) -> Result<WalkStream<T>>
926    where
927        T: 'static,
928    {
929        let ordering = self.inner.config.oid_ordering;
930        let max_results = self.inner.config.max_walk_results;
931        let walk_mode = self.inner.config.walk_mode;
932        let max_repetitions = max_repetitions_to_wire(self.inner.config.max_repetitions);
933        let version = self.inner.config.version;
934
935        WalkStream::new(
936            self.clone(),
937            oid,
938            version,
939            walk_mode,
940            ordering,
941            max_results,
942            max_repetitions,
943        )
944    }
945
946    /// Walk an OID subtree using GETNEXT.
947    ///
948    /// This method always uses GETNEXT regardless of the client's `WalkMode` configuration.
949    /// For auto-selection based on version and mode, use [`walk()`](Self::walk) instead.
950    ///
951    /// Returns an async stream that yields each variable binding in the subtree.
952    /// The walk terminates when an OID outside the subtree is encountered or
953    /// when `EndOfMibView` is returned.
954    ///
955    /// Uses the client's configured `oid_ordering` and `max_walk_results` settings.
956    ///
957    /// # Example
958    ///
959    /// ```rust,no_run
960    /// # use async_snmp::{Auth, Client, oid};
961    /// # async fn example() -> async_snmp::Result<()> {
962    /// # let client = Client::builder("127.0.0.1:161", Auth::v2c("public")).connect().await?;
963    /// // Force GETNEXT even for V2c/V3 clients
964    /// let results = client.walk_getnext(oid!(1, 3, 6, 1, 2, 1, 1)).collect().await?;
965    /// # Ok(())
966    /// # }
967    /// ```
968    #[instrument(skip(self), fields(snmp.target = %self.peer_addr(), snmp.oid = %oid))]
969    pub fn walk_getnext(&self, oid: Oid) -> Walk<T>
970    where
971        T: 'static,
972    {
973        let ordering = self.inner.config.oid_ordering;
974        let max_results = self.inner.config.max_walk_results;
975        Walk::new(self.clone(), oid, ordering, max_results)
976    }
977
978    /// Walk an OID subtree using GETBULK (more efficient than GETNEXT).
979    ///
980    /// Returns an async stream that yields each variable binding in the subtree.
981    /// Uses GETBULK internally with `non_repeaters=0`, fetching `max_repetitions`
982    /// values per request for efficient table traversal.
983    ///
984    /// Uses the client's configured `oid_ordering` and `max_walk_results` settings.
985    ///
986    /// # Arguments
987    ///
988    /// * `oid` - The base OID of the subtree to walk
989    /// * `max_repetitions` - How many OIDs to fetch per request
990    ///
991    /// # Example
992    ///
993    /// ```rust,no_run
994    /// # use async_snmp::{Auth, Client, oid};
995    /// # async fn example() -> async_snmp::Result<()> {
996    /// # let client = Client::builder("127.0.0.1:161", Auth::v2c("public")).connect().await?;
997    /// // Walk the interfaces table efficiently
998    /// let walk = client.bulk_walk(oid!(1, 3, 6, 1, 2, 1, 2, 2), 25);
999    /// // Process with futures StreamExt
1000    /// # Ok(())
1001    /// # }
1002    /// ```
1003    #[instrument(skip(self), fields(snmp.target = %self.peer_addr(), snmp.oid = %oid, snmp.max_repetitions = max_repetitions))]
1004    pub fn bulk_walk(&self, oid: Oid, max_repetitions: i32) -> BulkWalk<T>
1005    where
1006        T: 'static,
1007    {
1008        let ordering = self.inner.config.oid_ordering;
1009        let max_results = self.inner.config.max_walk_results;
1010        BulkWalk::new(self.clone(), oid, max_repetitions, ordering, max_results)
1011    }
1012
1013    /// Walk an OID subtree using the client's configured `max_repetitions`.
1014    ///
1015    /// This is a convenience method that uses the client's `max_repetitions` setting
1016    /// (default: 25) instead of requiring it as a parameter.
1017    ///
1018    /// # Example
1019    ///
1020    /// ```rust,no_run
1021    /// # use async_snmp::{Auth, Client, oid};
1022    /// # async fn example() -> async_snmp::Result<()> {
1023    /// # let client = Client::builder("127.0.0.1:161", Auth::v2c("public")).connect().await?;
1024    /// // Walk using configured max_repetitions
1025    /// let walk = client.bulk_walk_default(oid!(1, 3, 6, 1, 2, 1, 2, 2));
1026    /// // Process with futures StreamExt
1027    /// # Ok(())
1028    /// # }
1029    /// ```
1030    #[instrument(skip(self), fields(snmp.target = %self.peer_addr(), snmp.oid = %oid))]
1031    pub fn bulk_walk_default(&self, oid: Oid) -> BulkWalk<T>
1032    where
1033        T: 'static,
1034    {
1035        let ordering = self.inner.config.oid_ordering;
1036        let max_results = self.inner.config.max_walk_results;
1037        let max_repetitions = max_repetitions_to_wire(self.inner.config.max_repetitions);
1038        BulkWalk::new(self.clone(), oid, max_repetitions, ordering, max_results)
1039    }
1040}
1041
1042#[cfg(test)]
1043mod tests {
1044    use super::*;
1045    use crate::message::CommunityMessage;
1046    use crate::oid::Oid;
1047    use crate::pdu::{Pdu, PduType};
1048    use crate::varbind::VarBind;
1049    use crate::version::Version;
1050    use bytes::Bytes;
1051    use std::collections::VecDeque;
1052    use std::net::SocketAddr;
1053    use std::sync::{Arc, Mutex};
1054
1055    // -------------------------------------------------------------------------
1056    // max_repetitions_to_wire: saturating u32 -> i32 conversion used by
1057    // `walk()` and `bulk_walk_default()` so a configured `max_repetitions`
1058    // above `i32::MAX` cannot wrap to a negative GETBULK field (C7).
1059    // -------------------------------------------------------------------------
1060
1061    #[test]
1062    fn test_max_repetitions_to_wire_saturates_above_i32_max() {
1063        // A plain `as i32` cast would wrap this to a negative number.
1064        assert_eq!(max_repetitions_to_wire(u32::MAX), i32::MAX);
1065        assert_eq!(
1066            max_repetitions_to_wire(i32::MAX as u32 + 1),
1067            i32::MAX,
1068            "one past i32::MAX must saturate, not wrap negative"
1069        );
1070    }
1071
1072    #[test]
1073    fn test_max_repetitions_to_wire_passes_through_in_range_values() {
1074        assert_eq!(max_repetitions_to_wire(0), 0);
1075        assert_eq!(max_repetitions_to_wire(25), 25);
1076        assert_eq!(max_repetitions_to_wire(i32::MAX as u32), i32::MAX);
1077    }
1078
1079    // -------------------------------------------------------------------------
1080    // Mock transport that returns a response with a configurable number of
1081    // varbinds, regardless of how many were requested.
1082    // -------------------------------------------------------------------------
1083
1084    #[derive(Clone)]
1085    struct TruncatingTransport {
1086        /// Number of varbinds to include in each response.
1087        response_varbind_count: usize,
1088        /// Captured (`request_id`) values from sent requests, stored for building
1089        /// responses.
1090        pending: Arc<Mutex<VecDeque<i32>>>,
1091    }
1092
1093    impl TruncatingTransport {
1094        fn new(response_varbind_count: usize) -> Self {
1095            Self {
1096                response_varbind_count,
1097                pending: Arc::new(Mutex::new(VecDeque::new())),
1098            }
1099        }
1100    }
1101
1102    impl Transport for TruncatingTransport {
1103        fn send(&self, data: &[u8]) -> impl std::future::Future<Output = Result<()>> + Send {
1104            // Decode the sent request to extract the request_id.
1105            let request_id = crate::transport::extract_request_id(data).unwrap_or(1);
1106            {
1107                let mut q = self.pending.lock().unwrap();
1108                q.push_back(request_id);
1109            }
1110            async { Ok(()) }
1111        }
1112
1113        fn recv(
1114            &self,
1115            _request_id: i32,
1116        ) -> impl std::future::Future<Output = Result<(Bytes, SocketAddr)>> + Send {
1117            let request_id = {
1118                let mut q = self.pending.lock().unwrap();
1119                q.pop_front().unwrap_or(1)
1120            };
1121            let n = self.response_varbind_count;
1122            let peer: SocketAddr = "127.0.0.1:161".parse().unwrap();
1123
1124            async move {
1125                // Build a response PDU with n varbinds (NULL values).
1126                let varbinds: Vec<VarBind> = (0..n)
1127                    .map(|i| {
1128                        VarBind::new(
1129                            Oid::from_slice(&[1, 3, 6, 1, i as u32]),
1130                            crate::value::Value::Null,
1131                        )
1132                    })
1133                    .collect();
1134
1135                let pdu = Pdu {
1136                    pdu_type: PduType::Response,
1137                    request_id,
1138                    error_status: 0,
1139                    error_index: 0,
1140                    varbinds,
1141                };
1142
1143                let msg = CommunityMessage::v2c(Bytes::from_static(b"public"), pdu);
1144                let encoded = msg.encode();
1145                Ok((encoded, peer))
1146            }
1147        }
1148
1149        fn peer_addr(&self) -> SocketAddr {
1150            "127.0.0.1:161".parse().unwrap()
1151        }
1152
1153        fn local_addr(&self) -> SocketAddr {
1154            "127.0.0.1:0".parse().unwrap()
1155        }
1156
1157        fn is_reliable(&self) -> bool {
1158            true
1159        }
1160    }
1161
1162    fn make_client(response_varbind_count: usize) -> Client<TruncatingTransport> {
1163        let transport = TruncatingTransport::new(response_varbind_count);
1164        let config = ClientConfig {
1165            version: Version::V2c,
1166            max_oids_per_request: 10,
1167            retry: crate::client::retry::Retry::none(),
1168            ..Default::default()
1169        };
1170        Client::new(transport, config)
1171    }
1172
1173    #[tokio::test]
1174    async fn get_many_warns_on_truncated_response() {
1175        // Request 3 OIDs but the mock returns only 1 varbind - should warn and return what we got.
1176        let client = make_client(1);
1177        let oids = [
1178            Oid::from_slice(&[1, 3, 6, 1, 1]),
1179            Oid::from_slice(&[1, 3, 6, 1, 2]),
1180            Oid::from_slice(&[1, 3, 6, 1, 3]),
1181        ];
1182
1183        let result = client.get_many(&oids).await;
1184        assert!(result.is_ok(), "expected Ok, got: {:?}", result.err());
1185        assert_eq!(result.unwrap().len(), 1);
1186    }
1187
1188    #[tokio::test]
1189    async fn get_many_rejects_inflated_response() {
1190        // Request 3 OIDs but the mock returns 5 varbinds.
1191        let client = make_client(5);
1192        let oids = [
1193            Oid::from_slice(&[1, 3, 6, 1, 1]),
1194            Oid::from_slice(&[1, 3, 6, 1, 2]),
1195            Oid::from_slice(&[1, 3, 6, 1, 3]),
1196        ];
1197
1198        let err = client.get_many(&oids).await.unwrap_err();
1199        assert!(
1200            matches!(*err, Error::MalformedResponse { .. }),
1201            "expected MalformedResponse, got: {err}"
1202        );
1203    }
1204
1205    #[tokio::test]
1206    async fn get_many_accepts_correct_response_count() {
1207        // Request 3 OIDs and the mock returns exactly 3 varbinds.
1208        let client = make_client(3);
1209        let oids = [
1210            Oid::from_slice(&[1, 3, 6, 1, 1]),
1211            Oid::from_slice(&[1, 3, 6, 1, 2]),
1212            Oid::from_slice(&[1, 3, 6, 1, 3]),
1213        ];
1214
1215        let result = client.get_many(&oids).await;
1216        assert!(result.is_ok(), "expected Ok, got: {:?}", result.err());
1217        assert_eq!(result.unwrap().len(), 3);
1218    }
1219
1220    #[tokio::test]
1221    async fn get_next_many_warns_on_truncated_response() {
1222        // Request 3 OIDs but the mock returns only 1 varbind - should warn and return what we got.
1223        let client = make_client(1);
1224        let oids = [
1225            Oid::from_slice(&[1, 3, 6, 1, 1]),
1226            Oid::from_slice(&[1, 3, 6, 1, 2]),
1227            Oid::from_slice(&[1, 3, 6, 1, 3]),
1228        ];
1229
1230        let result = client.get_next_many(&oids).await;
1231        assert!(result.is_ok(), "expected Ok, got: {:?}", result.err());
1232        assert_eq!(result.unwrap().len(), 1);
1233    }
1234
1235    #[tokio::test]
1236    async fn get_next_many_rejects_inflated_response() {
1237        // Request 3 OIDs but the mock returns 5 varbinds.
1238        let client = make_client(5);
1239        let oids = [
1240            Oid::from_slice(&[1, 3, 6, 1, 1]),
1241            Oid::from_slice(&[1, 3, 6, 1, 2]),
1242            Oid::from_slice(&[1, 3, 6, 1, 3]),
1243        ];
1244
1245        let err = client.get_next_many(&oids).await.unwrap_err();
1246        assert!(
1247            matches!(*err, Error::MalformedResponse { .. }),
1248            "expected MalformedResponse, got: {err}"
1249        );
1250    }
1251
1252    #[tokio::test]
1253    async fn get_next_many_accepts_correct_response_count() {
1254        // Request 3 OIDs and the mock returns exactly 3 varbinds.
1255        let client = make_client(3);
1256        let oids = [
1257            Oid::from_slice(&[1, 3, 6, 1, 1]),
1258            Oid::from_slice(&[1, 3, 6, 1, 2]),
1259            Oid::from_slice(&[1, 3, 6, 1, 3]),
1260        ];
1261
1262        let result = client.get_next_many(&oids).await;
1263        assert!(result.is_ok(), "expected Ok, got: {:?}", result.err());
1264        assert_eq!(result.unwrap().len(), 3);
1265    }
1266
1267    #[tokio::test]
1268    async fn set_many_warns_on_truncated_response() {
1269        // Request 3 varbinds but the mock returns only 1 - should warn and return what we got.
1270        let client = make_client(1);
1271        let varbinds = [
1272            (
1273                Oid::from_slice(&[1, 3, 6, 1, 1]),
1274                crate::value::Value::Integer(1),
1275            ),
1276            (
1277                Oid::from_slice(&[1, 3, 6, 1, 2]),
1278                crate::value::Value::Integer(2),
1279            ),
1280            (
1281                Oid::from_slice(&[1, 3, 6, 1, 3]),
1282                crate::value::Value::Integer(3),
1283            ),
1284        ];
1285
1286        let result = client.set_many(&varbinds).await;
1287        assert!(result.is_ok(), "expected Ok, got: {:?}", result.err());
1288        assert_eq!(result.unwrap().len(), 1);
1289    }
1290
1291    #[tokio::test]
1292    async fn set_many_rejects_inflated_response() {
1293        // Request 3 varbinds but the mock returns 5.
1294        let client = make_client(5);
1295        let varbinds = [
1296            (
1297                Oid::from_slice(&[1, 3, 6, 1, 1]),
1298                crate::value::Value::Integer(1),
1299            ),
1300            (
1301                Oid::from_slice(&[1, 3, 6, 1, 2]),
1302                crate::value::Value::Integer(2),
1303            ),
1304            (
1305                Oid::from_slice(&[1, 3, 6, 1, 3]),
1306                crate::value::Value::Integer(3),
1307            ),
1308        ];
1309
1310        let err = client.set_many(&varbinds).await.unwrap_err();
1311        assert!(
1312            matches!(*err, Error::MalformedResponse { .. }),
1313            "expected MalformedResponse, got: {err}"
1314        );
1315    }
1316
1317    #[tokio::test]
1318    async fn set_many_accepts_correct_response_count() {
1319        // Request 3 varbinds and the mock returns exactly 3.
1320        let client = make_client(3);
1321        let varbinds = [
1322            (
1323                Oid::from_slice(&[1, 3, 6, 1, 1]),
1324                crate::value::Value::Integer(1),
1325            ),
1326            (
1327                Oid::from_slice(&[1, 3, 6, 1, 2]),
1328                crate::value::Value::Integer(2),
1329            ),
1330            (
1331                Oid::from_slice(&[1, 3, 6, 1, 3]),
1332                crate::value::Value::Integer(3),
1333            ),
1334        ];
1335
1336        let result = client.set_many(&varbinds).await;
1337        assert!(result.is_ok(), "expected Ok, got: {:?}", result.err());
1338        assert_eq!(result.unwrap().len(), 3);
1339    }
1340
1341    // -------------------------------------------------------------------------
1342    // Mock transport that returns tooBig when request exceeds a varbind threshold.
1343    // -------------------------------------------------------------------------
1344
1345    #[derive(Clone)]
1346    struct TooBigTransport {
1347        /// Max varbinds per request before returning tooBig.
1348        max_varbinds: usize,
1349        pending: Arc<Mutex<VecDeque<(i32, usize)>>>,
1350    }
1351
1352    impl TooBigTransport {
1353        fn new(max_varbinds: usize) -> Self {
1354            Self {
1355                max_varbinds,
1356                pending: Arc::new(Mutex::new(VecDeque::new())),
1357            }
1358        }
1359    }
1360
1361    impl Transport for TooBigTransport {
1362        fn send(&self, data: &[u8]) -> impl std::future::Future<Output = Result<()>> + Send {
1363            let request_id = crate::transport::extract_request_id(data).unwrap_or(1);
1364            // Decode the message to count varbinds
1365            let msg = CommunityMessage::decode(Bytes::copy_from_slice(data)).unwrap();
1366            let varbind_count = msg.pdu.standard().unwrap().varbinds.len();
1367            {
1368                let mut q = self.pending.lock().unwrap();
1369                q.push_back((request_id, varbind_count));
1370            }
1371            async { Ok(()) }
1372        }
1373
1374        fn recv(
1375            &self,
1376            _request_id: i32,
1377        ) -> impl std::future::Future<Output = Result<(Bytes, SocketAddr)>> + Send {
1378            let (request_id, varbind_count) = {
1379                let mut q = self.pending.lock().unwrap();
1380                q.pop_front().unwrap_or((1, 0))
1381            };
1382            let max = self.max_varbinds;
1383            let peer: SocketAddr = "127.0.0.1:161".parse().unwrap();
1384
1385            async move {
1386                let pdu = if varbind_count > max {
1387                    // Return tooBig with empty varbinds (per RFC 3416)
1388                    Pdu {
1389                        pdu_type: PduType::Response,
1390                        request_id,
1391                        error_status: ErrorStatus::TooBig.as_i32(),
1392                        error_index: 0,
1393                        varbinds: vec![],
1394                    }
1395                } else {
1396                    // Echo back one varbind per requested OID
1397                    let varbinds: Vec<VarBind> = (0..varbind_count)
1398                        .map(|i| {
1399                            VarBind::new(
1400                                Oid::from_slice(&[1, 3, 6, 1, i as u32]),
1401                                crate::value::Value::Integer(i as i32),
1402                            )
1403                        })
1404                        .collect();
1405                    Pdu {
1406                        pdu_type: PduType::Response,
1407                        request_id,
1408                        error_status: 0,
1409                        error_index: 0,
1410                        varbinds,
1411                    }
1412                };
1413
1414                let msg = CommunityMessage::v2c(Bytes::from_static(b"public"), pdu);
1415                Ok((msg.encode(), peer))
1416            }
1417        }
1418
1419        fn peer_addr(&self) -> SocketAddr {
1420            "127.0.0.1:161".parse().unwrap()
1421        }
1422
1423        fn local_addr(&self) -> SocketAddr {
1424            "127.0.0.1:0".parse().unwrap()
1425        }
1426
1427        fn is_reliable(&self) -> bool {
1428            true
1429        }
1430    }
1431
1432    #[tokio::test]
1433    async fn get_many_bisects_on_too_big() {
1434        // Agent can handle at most 3 varbinds per request. We ask for 8.
1435        // With max_oids_per_request=10, the initial batch is all 8 OIDs.
1436        // That triggers tooBig, so it bisects to 4+4, each of which still
1437        // triggers tooBig, then bisects to 2+2+2+2 which all succeed.
1438        let transport = TooBigTransport::new(3);
1439        let config = ClientConfig {
1440            version: Version::V2c,
1441            max_oids_per_request: 10,
1442            retry: crate::client::retry::Retry::none(),
1443            ..Default::default()
1444        };
1445        let client = Client::new(transport, config);
1446
1447        let oids: Vec<Oid> = (0..8u32)
1448            .map(|i| Oid::from_slice(&[1, 3, 6, 1, i]))
1449            .collect();
1450
1451        let result = client.get_many(&oids).await.unwrap();
1452        assert_eq!(result.len(), 8);
1453    }
1454
1455    #[tokio::test]
1456    async fn get_many_single_oid_too_big_is_unrecoverable() {
1457        // Agent returns tooBig even for a single OID - can't bisect further.
1458        let transport = TooBigTransport::new(0);
1459        let config = ClientConfig {
1460            version: Version::V2c,
1461            max_oids_per_request: 10,
1462            retry: crate::client::retry::Retry::none(),
1463            ..Default::default()
1464        };
1465        let client = Client::new(transport, config);
1466
1467        let oids = [Oid::from_slice(&[1, 3, 6, 1, 1])];
1468        let err = client.get_many(&oids).await.unwrap_err();
1469        assert!(
1470            matches!(
1471                &*err,
1472                Error::Snmp {
1473                    status: ErrorStatus::TooBig,
1474                    ..
1475                }
1476            ),
1477            "expected TooBig, got: {err}"
1478        );
1479    }
1480
1481    #[tokio::test]
1482    async fn get_next_many_bisects_on_too_big() {
1483        // Same as get_many test but for GETNEXT.
1484        let transport = TooBigTransport::new(3);
1485        let config = ClientConfig {
1486            version: Version::V2c,
1487            max_oids_per_request: 10,
1488            retry: crate::client::retry::Retry::none(),
1489            ..Default::default()
1490        };
1491        let client = Client::new(transport, config);
1492
1493        let oids: Vec<Oid> = (0..8u32)
1494            .map(|i| Oid::from_slice(&[1, 3, 6, 1, i]))
1495            .collect();
1496
1497        let result = client.get_next_many(&oids).await.unwrap();
1498        assert_eq!(result.len(), 8);
1499    }
1500
1501    // Batched path: get_many with more OIDs than max_per_request.
1502    #[tokio::test]
1503    async fn get_many_batched_warns_on_truncated_response() {
1504        // max_oids_per_request = 10, request 12 OIDs, mock returns 1 per batch.
1505        // Should warn and return 2 varbinds (1 per batch).
1506        let transport = TruncatingTransport::new(1);
1507        let config = ClientConfig {
1508            version: Version::V2c,
1509            max_oids_per_request: 10,
1510            retry: crate::client::retry::Retry::none(),
1511            ..Default::default()
1512        };
1513        let client = Client::new(transport, config);
1514
1515        let oids: Vec<Oid> = (0..12u32)
1516            .map(|i| Oid::from_slice(&[1, 3, 6, 1, i]))
1517            .collect();
1518
1519        let result = client.get_many(&oids).await;
1520        assert!(result.is_ok(), "expected Ok, got: {:?}", result.err());
1521        assert_eq!(result.unwrap().len(), 2); // 1 varbind per batch, 2 batches
1522    }
1523
1524    #[tokio::test]
1525    async fn get_many_batched_rejects_inflated_response() {
1526        // max_oids_per_request = 10, request 12 OIDs, mock returns 12 per batch.
1527        let transport = TruncatingTransport::new(12);
1528        let config = ClientConfig {
1529            version: Version::V2c,
1530            max_oids_per_request: 10,
1531            retry: crate::client::retry::Retry::none(),
1532            ..Default::default()
1533        };
1534        let client = Client::new(transport, config);
1535
1536        let oids: Vec<Oid> = (0..12u32)
1537            .map(|i| Oid::from_slice(&[1, 3, 6, 1, i]))
1538            .collect();
1539
1540        let err = client.get_many(&oids).await.unwrap_err();
1541        assert!(
1542            matches!(*err, Error::MalformedResponse { .. }),
1543            "expected MalformedResponse, got: {err}"
1544        );
1545    }
1546
1547    // -------------------------------------------------------------------------
1548    // Mock transport returning a response with a configurable PDU type,
1549    // community, and message version, for response-validation tests.
1550    // -------------------------------------------------------------------------
1551
1552    #[derive(Clone)]
1553    struct AdversarialTransport {
1554        pdu_type: PduType,
1555        community: &'static [u8],
1556        respond_as_v1: bool,
1557        pending: Arc<Mutex<VecDeque<i32>>>,
1558    }
1559
1560    impl AdversarialTransport {
1561        fn new(pdu_type: PduType, community: &'static [u8], respond_as_v1: bool) -> Self {
1562            Self {
1563                pdu_type,
1564                community,
1565                respond_as_v1,
1566                pending: Arc::new(Mutex::new(VecDeque::new())),
1567            }
1568        }
1569    }
1570
1571    impl Transport for AdversarialTransport {
1572        fn send(&self, data: &[u8]) -> impl std::future::Future<Output = Result<()>> + Send {
1573            let request_id = crate::transport::extract_request_id(data).unwrap_or(1);
1574            self.pending.lock().unwrap().push_back(request_id);
1575            async { Ok(()) }
1576        }
1577
1578        fn recv(
1579            &self,
1580            _request_id: i32,
1581        ) -> impl std::future::Future<Output = Result<(Bytes, SocketAddr)>> + Send {
1582            let request_id = self.pending.lock().unwrap().pop_front().unwrap_or(1);
1583            let peer: SocketAddr = "127.0.0.1:161".parse().unwrap();
1584            let pdu = Pdu {
1585                pdu_type: self.pdu_type,
1586                request_id,
1587                error_status: 0,
1588                error_index: 0,
1589                varbinds: vec![VarBind::new(
1590                    Oid::from_slice(&[1, 3, 6, 1, 1]),
1591                    crate::value::Value::Null,
1592                )],
1593            };
1594            let community = Bytes::from_static(self.community);
1595            let msg = if self.respond_as_v1 {
1596                CommunityMessage::v1(community, pdu)
1597            } else {
1598                CommunityMessage::v2c(community, pdu)
1599            };
1600            let encoded = msg.encode();
1601            async move { Ok((encoded, peer)) }
1602        }
1603
1604        fn peer_addr(&self) -> SocketAddr {
1605            "127.0.0.1:161".parse().unwrap()
1606        }
1607
1608        fn local_addr(&self) -> SocketAddr {
1609            "127.0.0.1:0".parse().unwrap()
1610        }
1611
1612        fn is_reliable(&self) -> bool {
1613            true
1614        }
1615    }
1616
1617    fn adversarial_client(
1618        pdu_type: PduType,
1619        community: &'static [u8],
1620        respond_as_v1: bool,
1621    ) -> Client<AdversarialTransport> {
1622        let transport = AdversarialTransport::new(pdu_type, community, respond_as_v1);
1623        let config = ClientConfig {
1624            version: Version::V2c,
1625            retry: crate::client::retry::Retry::none(),
1626            ..Default::default()
1627        };
1628        Client::new(transport, config)
1629    }
1630
1631    /// Control: the adversarial transport is otherwise well-formed, so a
1632    /// Response PDU with the sent community passes validation.
1633    #[tokio::test]
1634    async fn response_validation_accepts_well_formed_response() {
1635        let client = adversarial_client(PduType::Response, b"public", false);
1636        let result = client.get(&Oid::from_slice(&[1, 3, 6, 1, 1])).await;
1637        assert!(result.is_ok(), "expected Ok, got: {:?}", result.err());
1638    }
1639
1640    /// RFC 3416 Section 4.2: an echoed request-type PDU with a matching
1641    /// request-id is not a Response and must be rejected.
1642    #[tokio::test]
1643    async fn response_validation_rejects_echoed_request_pdu() {
1644        let client = adversarial_client(PduType::GetRequest, b"public", false);
1645        let err = client
1646            .get(&Oid::from_slice(&[1, 3, 6, 1, 1]))
1647            .await
1648            .unwrap_err();
1649        assert!(
1650            matches!(*err, Error::MalformedResponse { .. }),
1651            "expected MalformedResponse, got: {err}"
1652        );
1653    }
1654
1655    /// A response whose community differs from the one sent is rejected.
1656    #[tokio::test]
1657    async fn response_validation_accepts_community_mismatch() {
1658        let client = adversarial_client(PduType::Response, b"other", false);
1659        let result = client.get(&Oid::from_slice(&[1, 3, 6, 1, 1])).await;
1660        assert!(result.is_ok(), "expected Ok, got: {:?}", result.err());
1661    }
1662
1663    /// A v1 response to a v2c request is rejected (version mismatch).
1664    #[tokio::test]
1665    async fn response_validation_rejects_version_mismatch() {
1666        let client = adversarial_client(PduType::Response, b"public", true);
1667        let err = client
1668            .get(&Oid::from_slice(&[1, 3, 6, 1, 1]))
1669            .await
1670            .unwrap_err();
1671        assert!(
1672            matches!(*err, Error::MalformedResponse { .. }),
1673            "expected MalformedResponse, got: {err}"
1674        );
1675    }
1676}
async_snmp/client/mod.rs

async_snmp/client/
mod.rs
