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
This commit is contained in:
Goffi 2015-09-29 17:54:24 +02:00
commit 1f29d5695d

View file

@ -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)