|
36 | 36 | from pysnmp.hlapi.v3arch.asyncio import ( |
37 | 37 | cmdgen, |
38 | 38 | UdpTransportTarget, |
| 39 | + Udp6TransportTarget, |
39 | 40 | get_cmd, |
40 | 41 | walk_cmd, |
41 | 42 | SnmpEngine, |
@@ -327,30 +328,38 @@ def lookup_operstatus(int_operstatus): |
327 | 328 | return operstatus_options.get(int_operstatus, "") |
328 | 329 |
|
329 | 330 |
|
330 | | -def _create_transport_target(host, port=161, timeout=20): |
331 | | - """Create appropriate transport target based on address family.""" |
| 331 | +def is_ipv6_address(host): |
332 | 332 | try: |
333 | | - # Try to resolve the hostname and determine if it's IPv4 or IPv6 |
334 | | - addr_info = ipaddress.ip_address(host) |
335 | | - if addr_info.version == 6: |
336 | | - return cmdgen.Udp6TransportTarget((host, port), timeout=timeout) |
337 | | - else: |
338 | | - return cmdgen.UdpTransportTarget((host, port), timeout=timeout) |
| 333 | + return ipaddress.ip_address(host).version == 6 |
339 | 334 | except ValueError: |
340 | | - # If host is not a direct IP, try to resolve it |
341 | 335 | import socket |
342 | 336 | try: |
343 | 337 | # Try IPv4 first |
344 | 338 | socket.inet_pton(socket.AF_INET, socket.gethostbyname(host)) |
345 | | - return cmdgen.UdpTransportTarget((host, port), timeout=timeout) |
| 339 | + return False |
346 | 340 | except (socket.error, socket.gaierror): |
347 | 341 | try: |
348 | 342 | # Try IPv6 |
349 | 343 | socket.inet_pton(socket.AF_INET6, socket.getaddrinfo(host, None, socket.AF_INET6)[0][4][0]) |
350 | | - return cmdgen.Udp6TransportTarget((host, port), timeout=timeout) |
| 344 | + return True |
351 | 345 | except (socket.error, socket.gaierror): |
352 | | - # Default to IPv4 if resolution fails |
353 | | - return cmdgen.UdpTransportTarget((host, port), timeout=timeout) |
| 346 | + return False |
| 347 | + |
| 348 | + |
| 349 | +async def _create_transport_target_async(host, port=161, timeout=20): |
| 350 | + """Create appropriate transport target based on address family.""" |
| 351 | + if is_ipv6_address(host): |
| 352 | + return await Udp6TransportTarget.create((host, port), timeout=timeout) |
| 353 | + else: |
| 354 | + return await UdpTransportTarget.create((host, port), timeout=timeout) |
| 355 | + |
| 356 | + |
| 357 | +def _create_transport_target(host, port=161, timeout=20): |
| 358 | + """Create appropriate transport target based on address family.""" |
| 359 | + if is_ipv6_address(host): |
| 360 | + return cmdgen.Udp6TransportTarget((host, port), timeout=timeout) |
| 361 | + else: |
| 362 | + return cmdgen.UdpTransportTarget((host, port), timeout=timeout) |
354 | 363 |
|
355 | 364 |
|
356 | 365 | def decode_type(module, current_oid, val): |
@@ -1166,9 +1175,9 @@ def _init_auth(self): |
1166 | 1175 | ) |
1167 | 1176 |
|
1168 | 1177 | async def setup(self): |
1169 | | - self.transport = await UdpTransportTarget.create( |
1170 | | - (self.m_args['host'], 161), |
1171 | | - timeout=self.m_args['timeout'] |
| 1178 | + self.transport = await _create_transport_target_async( |
| 1179 | + self.m_args['host'], 161, |
| 1180 | + timeout=self.m_args['timeout'] |
1172 | 1181 | ) |
1173 | 1182 |
|
1174 | 1183 | async def _collect_system(self): |
|
0 commit comments