erebion/libervia-backend
Watch
1
0
Fork
You've already forked libervia-backend
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

code style reformatting using black

Browse source
This commit is contained in:
Goffi 2018-06-27 20:14:46 +02:00
commit 744cb0c503
1 changed files with 21394 additions and 10488 deletions
Download patch file Download diff file Expand all files Collapse all files

77
sat/plugins/plugin_misc_ip.py
View file

@ -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 wokkel import disco, iwokkel
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 @@ PLUGIN_INFO = {
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 @@ PARAMS = """
</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 @@ class IPPlugin(object):
# 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 @@ class IPPlugin(object):
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 @@ class IPPlugin(object):
@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 @@ class IPPlugin(object):
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 @@ class IPPlugin(object):
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 @@ class IPPlugin(object):
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 @@ class IPPlugin(object):
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 @@ class IPPlugin(object):
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 @@ class IPPlugin(object):
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(object):
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 []