yumoqing/apppublic
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

bugfix

Browse Source
This commit is contained in:
yumoqing 2022-03-09 12:36:24 +08:00
parent a85c2a2879
commit ed3072aa17
1 changed files with 47 additions and 48 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

95
appPublic/across_nat.py
View File

@ -1,6 +1,7 @@
from traceback import print_exc
from natpmp import NATPMP as pmp from natpmp import NATPMP as pmp
from aioupnp.upnp import UPnP import upnpy
from requests import get from requests import get
from .background import Background from .background import Background
@ -11,10 +12,18 @@ class AcrossNat(object):
self.pmp_supported = True self.pmp_supported = True
self.upnp_supported = True self.upnp_supported = True
self.init_pmp() self.init_pmp()
self.init_upnp()
async def init_upnp(self): def init_upnp(self):
if self.upnp is None: try:
self.upnp = await UPnP.discover() upnp = upnpy.UPnP()
igd = upnp.discover()[0]
s_names = [ n for n in igd.services.keys() if n.startswith('WANPPPConn') ]
self.upnp = igd.services[s_names[0]]
except Exception as e:
print(e)
print_exc()
self.upnp_supported = False
def init_pmp(self): def init_pmp(self):
try: try:
@ -22,71 +31,61 @@ class AcrossNat(object):
except pmp.NATPMPUnsupportedError: except pmp.NATPMPUnsupportedError:
self.pmp_supported = False self.pmp_supported = False
async def get_external_ip(self): def get_external_ip(self):
if self.pmp_supported: if self.pmp_supported:
self.external_ip = pmp.get_public_address() self.external_ip = pmp.get_public_address()
return self.external_ip return self.external_ip
if self.upnp_supported: if self.upnp_supported:
if self.upnp is None: x = self.upnp.GetExternalIPAddress()
await self.init_upnp() return x['NewExternalIPAddress']
return await self.upnp.get_external_ip()
try: try:
return get('https://api.ipify.org').text return get('https://api.ipify.org').text
except: except:
return get('https://ipapi.co/ip/').text return get('https://ipapi.co/ip/').text
async def upnp_map_port(self, inner_port, def upnp_check_external_port(self, eport, protocol='TCP'):
protocol='TCP', from_port=40003, ip=None, desc=None): try:
self.upnp.GetSpecificPortMappingEntry(NewExternalPort=eport,
NewProtocol=protocol,
NewRemoteHost='')
return True
except:
return False
def upnp_map_port(self, inner_port,
protocol='TCP', from_port=40003,
ip=None, desc='test'):
if self.upnp is None:
await self.init_upnp()
protocol = protocol.upper() protocol = protocol.upper()
if ip is None:
ip = self.upnp.lan_address
all_mappings = [i for i in await self.upnp.get_redirects()]
x = [ i for i in all_mappings if i.internal_port == inner_port \
and i.lan_address == ip \
and i.protocol == protocol ]
if len(x) > 0:
return x[0].external_port
occupied_ports = [ i.external_port for i in all_mappings if i.protocol == protocol ]
external_port = from_port external_port = from_port
while external_port < 52333: while external_port < 52333:
if external_port not in occupied_ports: if self.upnp_map_port_check(external_port, protocol=protocol):
break external_port += 1
external_port += 1 continue
self.upnp.AddPortMapping(NewRemoteHost='',
if external_port < 52333: NewExternalPort=external_port,
await self.upnp.add_port_mapping(external_port, NewProtocol=protocol,
protocol, NewInternalPort=inner_port,
inner_port, NewInternalClient=ip,
ip, NewEnabled=1,
desc or 'user added') NewPortMappingDescription=desc,
NewLeaseDuration=0
)
return external_port return external_port
return None return None
async def is_port_mapped(self, external_port, protocol='TCP'): def is_port_mapped(self, external_port, protocol='TCP'):
if self.upnp is None:
await self.init_upnp()
protocol = protocol.upper() protocol = protocol.upper()
if self.upnp_supported: if self.upnp_supported:
x = await self.upnp.get_specific_port_mapping(external_port, return self.upnp_map_port_check(external_port,
protocol) protocol=protocol)
if len(x) == 0:
return True
return False
raise Exception('not implemented') raise Exception('not implemented')
async def port_unmap(self, external_port, protocol='TCP'): def port_unmap(self, external_port, protocol='TCP'):
if self.upnp is None:
await self.init_upnp()
protocol = protocol.upper() protocol = protocol.upper()
if self.upnp_supported: if self.upnp_supported:
await self.upnp.delete_port_mapping(external_port, protocol) self.upnp.delete_port_mapping(external_port, protocol)
raise Exception('not implemented') raise Exception('not implemented')
def pmp_map_port(self, inner_port, protocol='TCP', from_port=40003): def pmp_map_port(self, inner_port, protocol='TCP', from_port=40003):
@ -98,9 +97,9 @@ class AcrossNat(object):
lifetime=999999999) lifetime=999999999)
return x.public_port return x.public_port
async def map_port(self, inner_port, protocol='tcp', from_port=40003, lan_ip=None, desc=None): def map_port(self, inner_port, protocol='tcp', from_port=40003, lan_ip=None, desc=None):
if self.pmp_supported: if self.pmp_supported:
return self.pmp_map_port(inner_port, protocol=protocol) return self.pmp_map_port(inner_port, protocol=protocol)
return await self.upnp_map_port( inner_port, protocol=protocol, ip=lan_ip, desc=desc) return self.upnp_map_port( inner_port, protocol=protocol, ip=lan_ip, desc=desc)