Mercurial > libervia-backend
diff sat/plugins/plugin_misc_ip.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 | 003b8b4b56a7 |
line wrap: on
line diff
--- a/sat/plugins/plugin_misc_ip.py Wed Jun 27 07:51:29 2018 +0200 +++ b/sat/plugins/plugin_misc_ip.py Wed Jun 27 20:14:46 2018 +0200 @@ -20,6 +20,7 @@ from sat.core.i18n import _, D_ from sat.core.constants import Const as C from sat.core.log import getLogger + log = getLogger(__name__) from sat.tools import xml_tools from twisted.web import client as webclient @@ -34,10 +35,13 @@ from twisted.words.protocols.jabber.xmlstream import XMPPHandler from twisted.words.protocols.jabber.error import StanzaError import urlparse + try: import netifaces except ImportError: - log.warning(u"netifaces is not available, it help discovering IPs, you can install it on https://pypi.python.org/pypi/netifaces") + log.warning( + u"netifaces is not available, it help discovering IPs, you can install it on https://pypi.python.org/pypi/netifaces" + ) netifaces = None @@ -50,26 +54,29 @@ C.PI_RECOMMENDATIONS: ["NAT-PORT"], C.PI_MAIN: "IPPlugin", C.PI_HANDLER: "yes", - C.PI_DESCRIPTION: _("""This plugin help to discover our external IP address.""") + C.PI_DESCRIPTION: _("""This plugin help to discover our external IP address."""), } # TODO: GET_IP_PAGE should be configurable in sat.conf -GET_IP_PAGE = "http://salut-a-toi.org/whereami/" # This page must only return external IP of the requester +GET_IP_PAGE = ( + "http://salut-a-toi.org/whereami/" +) # This page must only return external IP of the requester GET_IP_LABEL = D_(u"Allow external get IP") GET_IP_CATEGORY = "General" GET_IP_NAME = "allow_get_ip" GET_IP_CONFIRM_TITLE = D_(u"Confirm external site request") -GET_IP_CONFIRM = D_(u"""To facilitate data transfer, we need to contact a website. +GET_IP_CONFIRM = D_( + u"""To facilitate data transfer, we need to contact a website. A request will be done on {page} That means that administrators of {domain} can know that you use "{app_name}" and your IP Address. IP address is an identifier to locate you on Internet (similar to a phone number). Do you agree to do this request ? -""").format( - page = GET_IP_PAGE, - domain = urlparse.urlparse(GET_IP_PAGE).netloc, - app_name = C.APP_NAME) +""" +).format( + page=GET_IP_PAGE, domain=urlparse.urlparse(GET_IP_PAGE).netloc, app_name=C.APP_NAME +) NS_IP_CHECK = "urn:xmpp:sic:1" PARAMS = """ @@ -80,7 +87,9 @@ </category> </general> </params> - """.format(category=GET_IP_CATEGORY, name=GET_IP_NAME, label=GET_IP_LABEL) + """.format( + category=GET_IP_CATEGORY, name=GET_IP_NAME, label=GET_IP_LABEL +) class IPPlugin(object): @@ -94,7 +103,7 @@ # NAT-Port try: - self._nat = host.plugins['NAT-PORT'] + self._nat = host.plugins["NAT-PORT"] except KeyError: log.debug(u"NAT port plugin not available") self._nat = None @@ -118,16 +127,26 @@ if parameter is not set, a dialog is shown to use to get its confirmation, and parameted is set according to answer @return (defer.Deferred[bool]): True if external request is autorised """ - allow_get_ip = self.host.memory.params.getParamA(GET_IP_NAME, GET_IP_CATEGORY, use_default=False) + allow_get_ip = self.host.memory.params.getParamA( + GET_IP_NAME, GET_IP_CATEGORY, use_default=False + ) if allow_get_ip is None: # we don't have autorisation from user yet to use get_ip, we ask him def setParam(allowed): # FIXME: we need to use boolConst as setParam only manage str/unicode # need to be fixed when params will be refactored - self.host.memory.setParam(GET_IP_NAME, C.boolConst(allowed), GET_IP_CATEGORY) + self.host.memory.setParam( + GET_IP_NAME, C.boolConst(allowed), GET_IP_CATEGORY + ) return allowed - d = xml_tools.deferConfirm(self.host, _(GET_IP_CONFIRM), _(GET_IP_CONFIRM_TITLE), profile=client.profile) + + d = xml_tools.deferConfirm( + self.host, + _(GET_IP_CONFIRM), + _(GET_IP_CONFIRM_TITLE), + profile=client.profile, + ) d.addCallback(setParam) return d @@ -140,7 +159,7 @@ @param ip_addr(str): IP addresse @return (bool): True if addresse is acceptable """ - return not ip_addr.startswith('127.') + return not ip_addr.startswith("127.") def _insertFirst(self, addresses, ip_addr): """Insert ip_addr as first item in addresses @@ -164,9 +183,9 @@ url = urlparse.urlparse(ext_url) port = url.port if port is None: - if url.scheme=='http': + if url.scheme == "http": port = 80 - elif url.scheme=='https': + elif url.scheme == "https": port = 443 else: log.error(u"Unknown url scheme: {}".format(url.scheme)) @@ -175,6 +194,7 @@ log.error(u"Can't find url hostname for {}".format(GET_IP_PAGE)) point = endpoints.TCP4ClientEndpoint(reactor, url.hostname, port) + def gotConnection(p): local_ip = p.transport.getHost().host p.transport.loseConnection() @@ -196,7 +216,7 @@ if self._local_ip_cache is not None: defer.returnValue(self._local_ip_cache) addresses = [] - localhost = ['127.0.0.1'] + localhost = ["127.0.0.1"] # we first try our luck with netifaces if netifaces is not None: @@ -208,7 +228,7 @@ except KeyError: continue for data in inet_list: - addresse = data['addr'] + addresse = data["addr"] if self._filterAddresse(addresse): addresses.append(addresse) @@ -251,19 +271,20 @@ if self._external_ip_cache is not None: defer.returnValue(self._external_ip_cache) - # we first try with XEP-0279 ip_check = yield self.host.hasFeature(client, NS_IP_CHECK) if ip_check: log.debug(u"Server IP Check available, we use it to retrieve our IP") iq_elt = client.IQ("get") - iq_elt.addElement((NS_IP_CHECK, 'address')) + iq_elt.addElement((NS_IP_CHECK, "address")) try: result_elt = yield iq_elt.send() - address_elt = result_elt.elements(NS_IP_CHECK, 'address').next() - ip_elt = address_elt.elements(NS_IP_CHECK,'ip').next() + address_elt = result_elt.elements(NS_IP_CHECK, "address").next() + ip_elt = address_elt.elements(NS_IP_CHECK, "ip").next() except StopIteration: - log.warning(u"Server returned invalid result on XEP-0279 request, we ignore it") + log.warning( + u"Server returned invalid result on XEP-0279 request, we ignore it" + ) except StanzaError as e: log.warning(u"error while requesting ip to server: {}".format(e)) else: @@ -289,7 +310,11 @@ log.warning(u"Can't access Domain Name System") ip = None except web_error.Error as e: - log.warning(u"Error while retrieving IP on {url}: {message}".format(url=GET_IP_PAGE, message=e)) + log.warning( + u"Error while retrieving IP on {url}: {message}".format( + url=GET_IP_PAGE, message=e + ) + ) ip = None else: self._external_ip_cache = ip @@ -299,8 +324,8 @@ class IPPlugin_handler(XMPPHandler): implements(iwokkel.IDisco) - def getDiscoInfo(self, requestor, target, nodeIdentifier=''): + def getDiscoInfo(self, requestor, target, nodeIdentifier=""): return [disco.DiscoFeature(NS_IP_CHECK)] - def getDiscoItems(self, requestor, target, nodeIdentifier=''): + def getDiscoItems(self, requestor, target, nodeIdentifier=""): return []