From b2446e8a9841625abed2e0b67185751b7f17b208 Mon Sep 17 00:00:00 2001 From: elie Date: Sun, 5 Jul 2015 14:22:26 +0000 Subject: - original asynsock transport and AsyncsockDispatcher renamed into asyncore and AsyncoreDispatcher respectively to provide better hint to fellow devs on the underlying transport being used - backward compatibility preserved through dummy asynsock symbols --- pysnmp/carrier/asyncore/base.py | 89 ++++++++++++++++++ pysnmp/carrier/asyncore/dgram/base.py | 154 +++++++++++++++++++++++++++++++ pysnmp/carrier/asyncore/dgram/udp.py | 14 +++ pysnmp/carrier/asyncore/dgram/udp6.py | 34 +++++++ pysnmp/carrier/asyncore/dgram/unix.py | 50 ++++++++++ pysnmp/carrier/asyncore/dispatch.py | 42 +++++++++ pysnmp/entity/config.py | 2 +- pysnmp/entity/rfc3413/oneliner/target.py | 2 +- pysnmp/proto/secmod/rfc2576.py | 2 +- 9 files changed, 386 insertions(+), 3 deletions(-) create mode 100644 pysnmp/carrier/asyncore/base.py create mode 100644 pysnmp/carrier/asyncore/dgram/base.py create mode 100644 pysnmp/carrier/asyncore/dgram/udp.py create mode 100644 pysnmp/carrier/asyncore/dgram/udp6.py create mode 100644 pysnmp/carrier/asyncore/dgram/unix.py create mode 100644 pysnmp/carrier/asyncore/dispatch.py (limited to 'pysnmp') diff --git a/pysnmp/carrier/asyncore/base.py b/pysnmp/carrier/asyncore/base.py new file mode 100644 index 0000000..45e2fb2 --- /dev/null +++ b/pysnmp/carrier/asyncore/base.py @@ -0,0 +1,89 @@ +# Defines standard API to asyncore-based transport +import socket, sys +import asyncore +from pysnmp.carrier import error +from pysnmp.carrier.base import AbstractTransport +from pysnmp.carrier.asyncore.dispatch import AsyncoreDispatcher +from pysnmp import debug + +class AbstractSocketTransport(asyncore.dispatcher, AbstractTransport): + protoTransportDispatcher = AsyncoreDispatcher + sockFamily = sockType = None + retryCount = 0; retryInterval = 0 + bufferSize = 131070 + def __init__(self, sock=None, sockMap=None): + asyncore.dispatcher.__init__(self) + if sock is None: + if self.sockFamily is None: + raise error.CarrierError( + 'Address family %s not supported' % self.__class__.__name__ + ) + if self.sockType is None: + raise error.CarrierError( + 'Socket type %s not supported' % self.__class__.__name__ + ) + try: + sock = socket.socket(self.sockFamily, self.sockType) + except socket.error: + raise error.CarrierError('socket() failed: %s' % sys.exc_info()[1]) + + try: + for b in socket.SO_RCVBUF, socket.SO_SNDBUF: + bsize = sock.getsockopt(socket.SOL_SOCKET, b) + if bsize < self.bufferSize: + sock.setsockopt(socket.SOL_SOCKET, b, self.bufferSize) + debug.logger & debug.flagIO and debug.logger('%s: socket %d buffer size increased from %d to %d for buffer %d' % (self.__class__.__name__, sock.fileno(), bsize, self.bufferSize, b)) + except Exception: + debug.logger & debug.flagIO and debug.logger('%s: socket buffer size option mangling failure for buffer %d: %s' % (self.__class__.__name__, b, sys.exc_info()[1])) + + # The socket map is managed by the AsyncoreDispatcher on + # which this transport is registered. Here we just prepare + # socket and postpone transport registration at dispatcher + # till AsyncoreDispatcher invokes registerSocket() + + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + sock.setblocking(0) + self.set_socket(sock) + + def __hash__(self): return hash(self.socket) + + # The following two methods are part of base class so here we overwrite + # them to separate socket management from dispatcher registration tasks. + # These two are just for dispatcher registration. + def add_channel(self, map=None): + if map is not None: + map[self._fileno] = self + self.connected = True + + def del_channel(self, map=None): + if map is not None and self._fileno in map: + del map[self._fileno] + self.connected = False + + def registerSocket(self, sockMap=None): + self.add_channel(sockMap) + + def unregisterSocket(self, sockMap=None): + self.del_channel(sockMap) + + # Public API + + def openClientMode(self, iface=None): + raise error.CarrierError('Method not implemented') + + def openServerMode(self, iface=None): + raise error.CarrierError('Method not implemented') + + def sendMessage(self, outgoingMessage, transportAddress): + raise error.CarrierError('Method not implemented') + + def closeTransport(self): + AbstractTransport.closeTransport(self) + self.close() + + # asyncore API + def handle_close(self): raise error.CarrierError( + 'Transport unexpectedly closed' + ) + def handle_error(self): raise + diff --git a/pysnmp/carrier/asyncore/dgram/base.py b/pysnmp/carrier/asyncore/dgram/base.py new file mode 100644 index 0000000..5b87e70 --- /dev/null +++ b/pysnmp/carrier/asyncore/dgram/base.py @@ -0,0 +1,154 @@ +# Implements asyncore-based generic DGRAM transport +import socket, errno, sys +from pysnmp.carrier.asyncore.base import AbstractSocketTransport +from pysnmp.carrier import sockfix, sockmsg, error +from pysnmp import debug + +sockErrors = { # Ignore these socket errors + errno.ESHUTDOWN: 1, + errno.ENOTCONN: 1, + errno.ECONNRESET: 0, + errno.ECONNREFUSED: 0, + errno.EAGAIN: 0, + errno.EWOULDBLOCK: 0 + } +if hasattr(errno, 'EBADFD'): + # bad FD may happen upon FD closure on n-1 select() event + sockErrors[errno.EBADFD] = 1 + +class DgramSocketTransport(AbstractSocketTransport): + sockType = socket.SOCK_DGRAM + retryCount = 3; retryInterval = 1 + addressType = lambda x: x + def __init__(self, sock=None, sockMap=None): + self.__outQueue = [] + self._sendto = lambda s,b,a: s.sendto(b, a) + def __recvfrom(s, sz): + d, a = s.recvfrom(sz) + return d, self.addressType(a) + self._recvfrom = __recvfrom + AbstractSocketTransport.__init__(self, sock, sockMap) + + def openClientMode(self, iface=None): + if iface is not None: + try: + self.socket.bind(iface) + except socket.error: + raise error.CarrierError('bind() for %s failed: %s' % (iface is None and "" or iface, sys.exc_info()[1])) + return self + + def openServerMode(self, iface): + try: + self.socket.bind(iface) + except socket.error: + raise error.CarrierError('bind() for %s failed: %s' % (iface, sys.exc_info()[1],)) + return self + + def enableBroadcast(self, flag=1): + try: + self.socket.setsockopt( + socket.SOL_SOCKET, socket.SO_BROADCAST, flag + ) + except socket.error: + raise error.CarrierError('setsockopt() for SO_BROADCAST failed: %s' % (sys.exc_info()[1],)) + debug.logger & debug.flagIO and debug.logger('enableBroadcast: %s option SO_BROADCAST on socket %s' % (turnOn and "enabled" or "disabled", self.socket.fileno())) + return self + + def enablePktInfo(self, flag=1): + if not hasattr(self.socket, 'sendmsg') or \ + not hasattr(self.socket, 'recvmsg'): + raise error.CarrierError('sendmsg()/recvmsg() interface is not supported by this OS and/or Python version') + + try: + if self.socket.family in (socket.AF_INET,socket.AF_INET6): + self.socket.setsockopt(socket.SOL_IP, socket.IP_PKTINFO, flag) + if self.socket.family == socket.AF_INET6: + self.socket.setsockopt(socket.SOL_IPV6, socket.IPV6_RECVPKTINFO, flag) + except socket.error: + raise error.CarrierError('setsockopt() for %s failed: %s' % (self.socket.family == socket.AF_INET6 and "IPV6_RECVPKTINFO" or "IP_PKTINFO", sys.exc_info()[1])) + + self._sendto = sockmsg.getSendTo(self.addressType) + self._recvfrom = sockmsg.getRecvFrom(self.addressType) + + debug.logger & debug.flagIO and debug.logger('enablePktInfo: %s option %s on socket %s' % (self.socket.family == socket.AF_INET6 and "IPV6_RECVPKTINFO" or "IP_PKTINFO", flag and "enabled" or "disabled", self.socket.fileno())) + return self + + def enableTransparent(self, flag=1): + try: + if self.socket.family == socket.AF_INET: + self.socket.setsockopt( + socket.SOL_IP, socket.IP_TRANSPARENT, flag + ) + if self.socket.family == socket.AF_INET6: + self.socket.setsockopt( + socket.SOL_IPV6, socket.IP_TRANSPARENT, flag + ) + except socket.error: + raise error.CarrierError('setsockopt() for IP_TRANSPARENT failed: %s' % sys.exc_info()[1]) + except PermissionError: + raise error.CarrierError('IP_TRANSPARENT socket option requires superusre previleges') + + debug.logger & debug.flagIO and debug.logger('enableTransparent: %s option IP_TRANSPARENT on socket %s' % (flag and "enabled" or "disabled", self.socket.fileno())) + return self + + def sendMessage(self, outgoingMessage, transportAddress): + self.__outQueue.append( + (outgoingMessage, self.normalizeAddress(transportAddress)) + ) + debug.logger & debug.flagIO and debug.logger('sendMessage: outgoingMessage queued (%d octets) %s' % (len(outgoingMessage), debug.hexdump(outgoingMessage))) + + def normalizeAddress(self, transportAddress): + if not isinstance(transportAddress, self.addressType): + transportAddress = self.addressType(transportAddress) + if not transportAddress.getLocalAddress(): + transportAddress.setLocalAddress(self.getLocalAddress()) + return transportAddress + + def getLocalAddress(self): + # one evil OS does not seem to support getsockname() for DGRAM sockets + try: + return self.socket.getsockname() + except: + return ('0.0.0.0', 0) + + # asyncore API + def handle_connect(self): pass + def writable(self): return self.__outQueue + def handle_write(self): + outgoingMessage, transportAddress = self.__outQueue.pop(0) + debug.logger & debug.flagIO and debug.logger('handle_write: transportAddress %r -> %r outgoingMessage (%d octets) %s' % (transportAddress.getLocalAddress(), transportAddress, len(outgoingMessage), debug.hexdump(outgoingMessage))) + if not transportAddress: + debug.logger & debug.flagIO and debug.logger('handle_write: missing dst address, loosing outgoing msg') + return + try: + self._sendto( + self.socket, outgoingMessage, transportAddress + ) + except socket.error: + if sys.exc_info()[1].args[0] in sockErrors: + debug.logger & debug.flagIO and debug.logger('handle_write: ignoring socket error %s' % (sys.exc_info()[1],)) + else: + raise error.CarrierError('sendto() failed for %s: %s' % (transportAddress, sys.exc_info()[1])) + + def readable(self): return 1 + def handle_read(self): + try: + incomingMessage, transportAddress = self._recvfrom( + self.socket, 65535 + ) + transportAddress = self.normalizeAddress(transportAddress) + debug.logger & debug.flagIO and debug.logger('handle_read: transportAddress %r -> %r incomingMessage (%d octets) %s' % (transportAddress, transportAddress.getLocalAddress(), len(incomingMessage), debug.hexdump(incomingMessage))) + if not incomingMessage: + self.handle_close() + return + else: + self._cbFun(self, transportAddress, incomingMessage) + return + except socket.error: + if sys.exc_info()[1].args[0] in sockErrors: + debug.logger & debug.flagIO and debug.logger('handle_read: known socket error %s' % (sys.exc_info()[1],)) + sockErrors[sys.exc_info()[1].args[0]] and self.handle_close() + return + else: + raise error.CarrierError('recvfrom() failed: %s' % (sys.exc_info()[1],)) + def handle_close(self): pass # no datagram connection diff --git a/pysnmp/carrier/asyncore/dgram/udp.py b/pysnmp/carrier/asyncore/dgram/udp.py new file mode 100644 index 0000000..c1cf202 --- /dev/null +++ b/pysnmp/carrier/asyncore/dgram/udp.py @@ -0,0 +1,14 @@ +# Implements asyncore-based UDP transport domain +from socket import AF_INET +from pysnmp.carrier.base import AbstractTransportAddress +from pysnmp.carrier.asyncore.dgram.base import DgramSocketTransport + +domainName = snmpUDPDomain = (1, 3, 6, 1, 6, 1, 1) + +class UdpTransportAddress(tuple, AbstractTransportAddress): pass + +class UdpSocketTransport(DgramSocketTransport): + sockFamily = AF_INET + addressType = UdpTransportAddress + +UdpTransport = UdpSocketTransport diff --git a/pysnmp/carrier/asyncore/dgram/udp6.py b/pysnmp/carrier/asyncore/dgram/udp6.py new file mode 100644 index 0000000..73d0d66 --- /dev/null +++ b/pysnmp/carrier/asyncore/dgram/udp6.py @@ -0,0 +1,34 @@ +# Implements asyncore-based UDP6 transport domain +from pysnmp.carrier import sockfix +from pysnmp.carrier.base import AbstractTransportAddress +from pysnmp.carrier.asyncore.dgram.base import DgramSocketTransport +import socket + +domainName = snmpUDP6Domain = (1, 3, 6, 1, 2, 1, 100, 1, 2) + +class Udp6TransportAddress(tuple, AbstractTransportAddress): pass + +class Udp6SocketTransport(DgramSocketTransport): + sockFamily = socket.has_ipv6 and socket.AF_INET6 or None + addressType = Udp6TransportAddress + + def normalizeAddress(self, transportAddress): + if '%' in transportAddress[0]: # strip zone ID + ta = self.addressType( + (transportAddress[0].split('%')[0], + transportAddress[1], + 0, # flowinfo + 0) # scopeid + ) + else: + ta = self.addressType( + (transportAddress[0], transportAddress[1], 0, 0) + ) + + if isinstance(transportAddress, self.addressType) and \ + transportAddress.getLocalAddress(): + return ta.setLocalAddress(transportAddress.getLocalAddress()) + else: + return ta.setLocalAddress(self.getLocalAddress()) + +Udp6Transport = Udp6SocketTransport diff --git a/pysnmp/carrier/asyncore/dgram/unix.py b/pysnmp/carrier/asyncore/dgram/unix.py new file mode 100644 index 0000000..de53d44 --- /dev/null +++ b/pysnmp/carrier/asyncore/dgram/unix.py @@ -0,0 +1,50 @@ +# Implements asyncore-based UNIX transport domain +import os +import random +try: + from socket import AF_UNIX +except ImportError: + AF_UNIX = None +from pysnmp.carrier.base import AbstractTransportAddress +from pysnmp.carrier.asyncore.dgram.base import DgramSocketTransport + +domainName = snmpLocalDomain = (1, 3, 6, 1, 2, 1, 100, 1, 13) + +random.seed() + +class UnixTransportAddress(str, AbstractTransportAddress): pass + +class UnixSocketTransport(DgramSocketTransport): + sockFamily = AF_UNIX + addressType = UnixTransportAddress + + def openClientMode(self, iface=None): + if iface is None: + # UNIX domain sockets must be explicitly bound + iface = '' + while len(iface) < 8: + iface += chr(random.randrange(65, 91)) + iface += chr(random.randrange(97, 123)) + iface = os.path.sep + 'tmp' + os.path.sep + 'pysnmp' + iface + if os.path.exists(iface): + os.remove(iface) + DgramSocketTransport.openClientMode(self, iface) + self.__iface = iface + return self + + def openServerMode(self, iface): + DgramSocketTransport.openServerMode(self, iface) + self.__iface = iface + return self + + def closeTransport(self): + DgramSocketTransport.closeTransport(self) + try: + os.remove(self.__iface) + except OSError: + pass + +UnixTransport = UnixSocketTransport + +# Compatibility stub +UnixDgramSocketTransport = UnixSocketTransport diff --git a/pysnmp/carrier/asyncore/dispatch.py b/pysnmp/carrier/asyncore/dispatch.py new file mode 100644 index 0000000..790152d --- /dev/null +++ b/pysnmp/carrier/asyncore/dispatch.py @@ -0,0 +1,42 @@ +# Implements I/O over asynchronous sockets +from time import time +from sys import exc_info +from traceback import format_exception +from asyncore import socket_map +from asyncore import loop +from pysnmp.carrier.base import AbstractTransportDispatcher +from pysnmp.error import PySnmpError + +class AsyncoreDispatcher(AbstractTransportDispatcher): + def __init__(self): + self.__sockMap = {} # use own map for MT safety + self.timeout = 0.5 + AbstractTransportDispatcher.__init__(self) + + def getSocketMap(self): return self.__sockMap + def setSocketMap(self, sockMap=socket_map): self.__sockMap = sockMap + + def registerTransport(self, tDomain, t): + AbstractTransportDispatcher.registerTransport(self, tDomain, t) + t.registerSocket(self.__sockMap) + + def unregisterTransport(self, tDomain): + self.getTransport(tDomain).unregisterSocket(self.__sockMap) + AbstractTransportDispatcher.unregisterTransport(self, tDomain) + + def transportsAreWorking(self): + for transport in self.__sockMap.values(): + if transport.writable(): + return 1 + return 0 + + def runDispatcher(self, timeout=0.0): + while self.jobsArePending() or self.transportsAreWorking(): + try: + loop(timeout and timeout or self.timeout, + use_poll=True, map=self.__sockMap, count=1) + except KeyboardInterrupt: + raise + except: + raise PySnmpError('poll error: %s' % ';'.join(format_exception(*exc_info()))) + self.handleTimerTick(time()) diff --git a/pysnmp/entity/config.py b/pysnmp/entity/config.py index 085eea5..5c1bd83 100644 --- a/pysnmp/entity/config.py +++ b/pysnmp/entity/config.py @@ -1,7 +1,7 @@ # Initial SNMP engine configuration functions. During further operation, # SNMP engine might be configured remotely (through SNMP). from pyasn1.compat.octets import null -from pysnmp.carrier.asynsock.dgram import udp, udp6, unix +from pysnmp.carrier.asyncore.dgram import udp, udp6, unix from pysnmp.proto.secmod.rfc3414.auth import hmacmd5, hmacsha, noauth from pysnmp.proto.secmod.rfc3414.priv import des, nopriv from pysnmp.proto.secmod.rfc3826.priv import aes diff --git a/pysnmp/entity/rfc3413/oneliner/target.py b/pysnmp/entity/rfc3413/oneliner/target.py index 447fcc8..04ecb5f 100644 --- a/pysnmp/entity/rfc3413/oneliner/target.py +++ b/pysnmp/entity/rfc3413/oneliner/target.py @@ -1,5 +1,5 @@ import socket, sys -from pysnmp.carrier.asynsock.dgram import udp, udp6, unix +from pysnmp.carrier.asyncore.dgram import udp, udp6, unix from pysnmp import error from pyasn1.compat.octets import null diff --git a/pysnmp/proto/secmod/rfc2576.py b/pysnmp/proto/secmod/rfc2576.py index 7210ad8..6a7d1e9 100644 --- a/pysnmp/proto/secmod/rfc2576.py +++ b/pysnmp/proto/secmod/rfc2576.py @@ -3,7 +3,7 @@ import sys from pyasn1.codec.ber import encoder from pyasn1.error import PyAsn1Error from pysnmp.proto.secmod import base -from pysnmp.carrier.asynsock.dgram import udp, udp6, unix +from pysnmp.carrier.asyncore.dgram import udp, udp6, unix from pysnmp.smi.error import NoSuchInstanceError from pysnmp.proto import errind, error from pysnmp import debug -- cgit v1.2.1