Mercurial > libervia-backend
changeset 1536:04a13d9ae265
plugin nat-port: NAT port first draft:
- this plugin use UPnP-IGD (and maybe later NAT-PMP) to open a port or get local/external IP
- for now, only IP retrieval is implemented
author | Goffi <goffi@goffi.org> |
---|---|
date | Tue, 29 Sep 2015 17:54:24 +0200 (2015-09-29) |
parents | c9ef16de3f13 |
children | 6fa9e8c02c34 |
files | src/plugins/plugin_misc_nat-port.py |
diffstat | 1 files changed, 84 insertions(+), 0 deletions(-) [+] |
line wrap: on
line diff
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/src/plugins/plugin_misc_nat-port.py Tue Sep 29 17:54:24 2015 +0200 @@ -0,0 +1,84 @@ +#!/usr/bin/python +# -*- coding: utf-8 -*- + +# SAT plugin for NAT port mapping +# Copyright (C) 2009, 2010, 2011, 2012, 2013, 2014, 2015 Jérôme Poisson (goffi@goffi.org) + +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. + +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Affero General Public License for more details. + +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see <http://www.gnu.org/licenses/>. + +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 +from twisted.internet import defer + +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/") + + +PLUGIN_INFO = { + "name": "NAT port mapping", + "import_name": "NAT-PORT", + "type": C.PLUG_TYPE_MISC, + "main": "NatPort", + "handler": "no", + "description": _("""Automatic NAT port mapping using UPnP"""), +} + + +class NatPort(object): + # TODO: refresh data if a new connection is detected (see plugin_misc_ip) + + def __init__(self, host): + log.info(_("plugin NAT Port initialization")) + 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 + discover_d = threads.deferToThread(self._discover) + discover_d.chainDeferred(self._initialised) + self._initialised.addErrback(self._init_failed) + + def _init_failed(self, failure): + log.info(u"UPnP-GID not available") + self._upnp = None + + def _discover(self): + devices = self._upnp.discover() + if devices: + log.info(u"{nb} UPnP-IGD device(s) found".format(nb=devices)) + else: + log.info(u"Can't find UPnP-IGD device on the local network") + raise exceptions.NotFound + self._upnp.selectigd() + self._external_ip = self._upnp.externalipaddress() + + def getIP(self, local=False): + """Return IP address found with UPnP-IGD + + @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)