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

async_snmp/client/
mod.rs
