LBRYCommunity/aioupnp
1
0
Fork 0
Code Issues 7 Pull requests 1 Projects Releases 1 Packages Wiki Activity

fix miniupnpc fallback

Browse source
This commit is contained in:
Jack Robison 2018-08-07 13:53:35 -04:00
parent 64ad46ad22
commit 78e81a1dff
No known key found for this signature in database
GPG key ID: DF25C68FE0239BB2
4 changed files with 40 additions and 40 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
txupnp/__init__.py
View file

@ -1,14 +1,6 @@
__version__ = "0.0.1a6"
__version__ = "0.0.1a7"
__name__ = "txupnp"
__author__ = "Jack Robison"
__maintainer__ = "Jack Robison"
__license__ = "MIT"
__email__ = "jackrobison@lbry.io"
import logging
log = logging.getLogger(__name__)
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter('%(asctime)-15s-%(filename)s:%(lineno)s->%(message)s'))
log.addHandler(handler)
log.setLevel(logging.INFO)

19
txupnp/cli.py
View file

@ -2,7 +2,6 @@ import argparse
import logging
from twisted.internet import reactor, defer
from txupnp.upnp import UPnP
from txupnp.fault import UPnPError
log = logging.getLogger("txupnp")
@ -34,10 +33,9 @@ def add_mapping(u, *_):
port = 4567
protocol = "UDP"
description = "txupnp test mapping"
yield u.get_next_mapping(port, protocol, description)
result = yield u.get_specific_port_mapping(port, protocol)
if result:
print("added mapping")
ext_port = yield u.get_next_mapping(port, protocol, description)
if ext_port:
print("external port: %i to local %i/%s" % (ext_port, port, protocol))
@defer.inlineCallbacks
@ -74,6 +72,13 @@ def run_command(found, u, command, debug_xml):
def main():
import logging
log = logging.getLogger("txupnp")
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter('%(asctime)-15s-%(filename)s:%(lineno)s->%(message)s'))
log.addHandler(handler)
log.setLevel(logging.INFO)
parser = argparse.ArgumentParser(description="upnp command line utility")
parser.add_argument(dest="command", type=str, help="debug_gateway | list_mappings | get_external_ip | add_mapping | delete_mapping")
parser.add_argument("--debug_logging", dest="debug_logging", default=False, action="store_true")
@ -85,6 +90,10 @@ def main():
observer.start()
log.setLevel(logging.DEBUG)
command = args.command
command = command.replace("-", "_")
if command not in cli_commands:
print("unrecognized command: %s is not in %s" % (command, cli_commands.keys()))
return
def show(err):
print("error: {}".format(err))

10
txupnp/scpd.py
View file

@ -109,7 +109,7 @@ class _SCPDCommand(object):
@defer.inlineCallbacks
def __call__(self, **kwargs):
if set(kwargs.keys()) != set(self.param_names):
raise Exception("argument mismatch")
raise Exception("argument mismatch: %s vs %s" % (kwargs.keys(), self.param_names))
response = yield self.send_upnp_soap(**kwargs)
try:
result = self._process_result(response)
@ -221,7 +221,7 @@ class SCPDCommandRunner(object):
@staticmethod
@return_types(none)
def AddPortMapping(NewRemoteHost, NewExternalPort, NewProtocol, NewInternalPort, NewInternalClient,
NewEnabled, NewPortMappingDescription, NewLeaseDuration):
NewEnabled, NewPortMappingDescription, NewLeaseDuration=''):
"""Returns None"""
raise NotImplementedError()
@ -373,12 +373,12 @@ class UPnPFallback(object):
@return_types(none)
def AddPortMapping(self, NewRemoteHost, NewExternalPort, NewProtocol, NewInternalPort, NewInternalClient,
NewEnabled, NewPortMappingDescription, NewLeaseDuration):
NewEnabled, NewPortMappingDescription, NewLeaseDuration=''):
"""Returns None"""
if not self.available:
raise NotImplementedError()
return threads.deferToThread(self._upnp.addportmapping, NewExternalPort, NewProtocol, NewInternalPort,
NewInternalClient, NewPortMappingDescription, NewLeaseDuration)
return threads.deferToThread(self._upnp.addportmapping, NewExternalPort, NewProtocol, NewInternalClient,
NewInternalPort, NewPortMappingDescription, NewLeaseDuration)
def GetNATRSIPStatus(self):
"""Returns (NewRSIPAvailable, NewNATEnabled)"""

41
txupnp/upnp.py
View file

@ -1,6 +1,5 @@
import logging
import json
import treq
from twisted.internet import defer
from txupnp.fault import UPnPError
from txupnp.soap import SOAPServiceManager
@ -62,7 +61,7 @@ class UPnP(object):
@defer.inlineCallbacks
def start_miniupnpc_fallback(self):
found = False
if not self.commands and not self.miniupnpc_runner:
if not self.miniupnpc_runner:
log.debug("trying miniupnpc fallback")
fallback = UPnPFallback()
success = yield fallback.discover()
@ -78,12 +77,11 @@ class UPnP(object):
def get_external_ip(self):
return self.commands.GetExternalIPAddress()
def add_port_mapping(self, external_port, protocol, internal_port, lan_address, description, lease_duration):
def add_port_mapping(self, external_port, protocol, internal_port, lan_address, description):
return self.commands.AddPortMapping(
NewRemoteHost="", NewExternalPort=external_port, NewProtocol=protocol,
NewInternalPort=internal_port, NewInternalClient=lan_address,
NewEnabled=1, NewPortMappingDescription=description,
NewLeaseDuration=lease_duration
NewEnabled=1, NewPortMappingDescription=description, NewLeaseDuration=""
)
def get_port_mapping_by_index(self, index):
@ -118,7 +116,8 @@ class UPnP(object):
except UPnPError as err:
if 'NoSuchEntryInArray' in str(err):
defer.returnValue(None)
raise err
else:
raise err
def delete_port_mapping(self, external_port, protocol, new_remote_host=""):
"""
@ -149,23 +148,23 @@ class UPnP(object):
return self.commands.GetConnectionTypeInfo()
@defer.inlineCallbacks
def get_next_mapping(self, port, protocol, description):
def get_next_mapping(self, port, protocol, description, internal_port=None):
if protocol not in ["UDP", "TCP"]:
raise UPnPError("unsupported protocol: {}".format(protocol))
mappings = yield DeferredDict({p: self.get_specific_port_mapping(port, p)
for p in ["UDP", "TCP"]})
if not any((m is not None for m in mappings.values())): # there are no redirects for this port
yield self.add_port_mapping( # set one up
port, protocol, port, self.lan_address, description, 0
)
defer.returnValue(port)
if mappings[protocol]:
mapped_port = mappings[protocol][0]
mapped_address = mappings[protocol][1]
if mapped_port == port and mapped_address == self.lan_address: # reuse redirect to us
defer.returnValue(port)
port = yield self.get_next_mapping( # try the next port
port + 1, protocol, description
internal_port = internal_port or port
redirect_tups = yield self.get_redirects()
redirects = {
"%i:%s" % (ext_port, proto): (int_host, int_port, desc)
for (ext_host, ext_port, proto, int_port, int_host, enabled, desc, lease) in redirect_tups
}
while ("%i:%s" % (port, protocol)) in redirects:
int_host, int_port, _ = redirects["%i:%s" % (port, protocol)]
if int_host == self.lan_address and int_port == internal_port:
break
port += 1
yield self.add_port_mapping( # set one up
port, protocol, internal_port, self.lan_address, description
)
defer.returnValue(port)