Mercurial > libervia-backend
diff sat/plugins/plugin_misc_nat-port.py @ 2624:56f94936df1e
code style reformatting using black
author | Goffi <goffi@goffi.org> |
---|---|
date | Wed, 27 Jun 2018 20:14:46 +0200 |
parents | 26edcf3a30eb |
children | 378188abe941 |
line wrap: on
line diff
--- a/sat/plugins/plugin_misc_nat-port.py Wed Jun 27 07:51:29 2018 +0200 +++ b/sat/plugins/plugin_misc_nat-port.py Wed Jun 27 20:14:46 2018 +0200 @@ -20,6 +20,7 @@ from sat.core.i18n import _ from sat.core.constants import Const as C from sat.core.log import getLogger + log = getLogger(__name__) from sat.core import exceptions from twisted.internet import threads @@ -30,7 +31,9 @@ try: import miniupnpc except ImportError: - raise exceptions.MissingModule(u"Missing module MiniUPnPc, please download/install it (and its Python binding) at http://miniupnp.free.fr/ (or use pip install miniupnpc)") + raise exceptions.MissingModule( + u"Missing module MiniUPnPc, please download/install it (and its Python binding) at http://miniupnp.free.fr/ (or use pip install miniupnpc)" + ) PLUGIN_INFO = { @@ -42,8 +45,10 @@ C.PI_DESCRIPTION: _("""Automatic NAT port mapping using UPnP"""), } -STARTING_PORT = 6000 # starting point to automatically find a port -DEFAULT_DESC = u'SaT port mapping' # we don't use "à" here as some bugged NAT don't manage charset correctly +STARTING_PORT = 6000 # starting point to automatically find a port +DEFAULT_DESC = ( + u"SaT port mapping" +) # we don't use "à" here as some bugged NAT don't manage charset correctly class MappingError(Exception): @@ -58,11 +63,11 @@ self.host = host self._external_ip = None self._initialised = defer.Deferred() - self._upnp = miniupnpc.UPnP() # will be None if no device is available - self._upnp.discoverdelay=200 - self._mutex = threading.Lock() # used to protect access to self._upnp - self._starting_port_cache = None # used to cache the first available port - self._to_unmap = [] # list of tuples (ext_port, protocol) of ports to unmap on unload + self._upnp = miniupnpc.UPnP() # will be None if no device is available + self._upnp.discoverdelay = 200 + self._mutex = threading.Lock() # used to protect access to self._upnp + self._starting_port_cache = None # used to cache the first available port + self._to_unmap = [] # list of tuples (ext_port, protocol) of ports to unmap on unload discover_d = threads.deferToThread(self._discover) discover_d.chainDeferred(self._initialised) self._initialised.addErrback(self._init_failed) @@ -99,12 +104,14 @@ @param local(bool): True to get external IP address, False to get local network one @return (None, str): found IP address, or None of something got wrong """ + def getIP(dummy): if self._upnp is None: return None # lanaddr can be the empty string if not found, # we need to return None in this case return (self._upnp.lanaddr or None) if local else self._external_ip + return self._initialised.addCallback(getIP) def _unmapPortsBlocking(self): @@ -115,11 +122,17 @@ log.info(u"Unmapping port {}".format(port)) unmapping = self._upnp.deleteportmapping( # the last parameter is remoteHost, we don't use it - port, protocol, '') + port, + protocol, + "", + ) if not unmapping: - log.error(u"Can't unmap port {port} ({protocol})".format( - port=port, protocol=protocol)) + log.error( + u"Can't unmap port {port} ({protocol})".format( + port=port, protocol=protocol + ) + ) del self._to_unmap[:] finally: self._mutex.release() @@ -143,8 +156,8 @@ ext_port = STARTING_PORT if starting_port is None else starting_port ret = self._upnp.getspecificportmapping(ext_port, protocol) while ret != None and ext_port < 65536: - ext_port += 1 - ret = self._upnp.getspecificportmapping(ext_port, protocol) + ext_port += 1 + ret = self._upnp.getspecificportmapping(ext_port, protocol) if starting_port is None: # XXX: we cache the first successfuly found external port # to avoid testing again the first series the next time @@ -153,7 +166,13 @@ try: mapping = self._upnp.addportmapping( # the last parameter is remoteHost, we don't use it - ext_port, protocol, self._upnp.lanaddr, int_port, desc, '') + ext_port, + protocol, + self._upnp.lanaddr, + int_port, + desc, + "", + ) except Exception as e: log.error(_(u"addportmapping error: {msg}").format(msg=e)) raise failure.Failure(MappingError()) @@ -167,7 +186,7 @@ return ext_port - def mapPort(self, int_port, ext_port=None, protocol='TCP', desc=DEFAULT_DESC): + def mapPort(self, int_port, ext_port=None, protocol="TCP", desc=DEFAULT_DESC): """Add a port mapping @param int_port(int): internal port to use @@ -179,19 +198,25 @@ """ if self._upnp is None: return defer.succeed(None) + def mappingCb(ext_port): - log.info(u"{protocol} mapping from {int_port} to {ext_port} successful".format( - protocol = protocol, - int_port = int_port, - ext_port = ext_port, - )) + log.info( + u"{protocol} mapping from {int_port} to {ext_port} successful".format( + protocol=protocol, int_port=int_port, ext_port=ext_port + ) + ) return ext_port + def mappingEb(failure_): failure_.trap(MappingError) log.warning(u"Can't map internal {int_port}".format(int_port=int_port)) + def mappingUnknownEb(failure_): log.error(_(u"error while trying to map ports: {msg}").format(msg=failure_)) - d = threads.deferToThread(self._mapPortBlocking, int_port, ext_port, protocol, desc) + + d = threads.deferToThread( + self._mapPortBlocking, int_port, ext_port, protocol, desc + ) d.addCallbacks(mappingCb, mappingEb) d.addErrback(mappingUnknownEb) return d