From c07ceea3a287ed231f4eb0e31e77b4e7504389b5 Mon Sep 17 00:00:00 2001 From: yumoqing Date: Fri, 22 Apr 2022 13:23:43 +0800 Subject: [PATCH] bugfix --- appPublic/across_nat.bak.py | 106 ++++++++++++++++++++++++ appPublic/dictObject.old.py | 161 ++++++++++++++++++++++++++++++++++++ 2 files changed, 267 insertions(+) create mode 100644 appPublic/across_nat.bak.py create mode 100644 appPublic/dictObject.old.py diff --git a/appPublic/across_nat.bak.py b/appPublic/across_nat.bak.py new file mode 100644 index 0000000..2871f73 --- /dev/null +++ b/appPublic/across_nat.bak.py @@ -0,0 +1,106 @@ + +from natpmp import NATPMP as pmp +from aioupnp.upnp import UPnP +from requests import get +from .background import Background + +class AcrossNat(object): + def __init__(self): + self.external_ip = None + self.upnp = None + self.pmp_supported = True + self.upnp_supported = True + self.init_pmp() + + async def init_upnp(self): + if self.upnp is None: + self.upnp = await UPnP.discover() + + def init_pmp(self): + try: + self.external_ip = pmp.get_public_address() + except pmp.NATPMPUnsupportedError: + self.pmp_supported = False + + async def get_external_ip(self): + if self.pmp_supported: + self.external_ip = pmp.get_public_address() + return self.external_ip + + if self.upnp_supported: + if self.upnp is None: + await self.init_upnp() + return await self.upnp.get_external_ip() + + try: + return get('https://api.ipify.org').text + except: + return get('https://ipapi.co/ip/').text + + async def upnp_map_port(self, inner_port, + protocol='TCP', from_port=40003, ip=None, desc=None): + + if self.upnp is None: + await self.init_upnp() + 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 + while external_port < 52333: + if external_port not in occupied_ports: + break + external_port += 1 + + if external_port < 52333: + await self.upnp.add_port_mapping(external_port, + protocol, + inner_port, + ip, + desc or 'user added') + return external_port + return None + + async def is_port_mapped(self, external_port, protocol='TCP'): + if self.upnp is None: + await self.init_upnp() + protocol = protocol.upper() + if self.upnp_supported: + x = await self.upnp.get_specific_port_mapping(external_port, + protocol) + if len(x) == 0: + return True + return False + raise Exception('not implemented') + + async def port_unmap(self, external_port, protocol='TCP'): + if self.upnp is None: + await self.init_upnp() + protocol = protocol.upper() + if self.upnp_supported: + await self.upnp.delete_port_mapping(external_port, protocol) + raise Exception('not implemented') + + def pmp_map_port(self, inner_port, protocol='TCP', from_port=40003): + if protocol.upper() == 'TCP': + x = pmp.map_tcp_port(from_port, inner_port, + lifetime=999999999) + return x.public_port + x = pmp.map_udp_port(from_port, inner_port, + lifetime=999999999) + return x.public_port + + async def map_port(self, inner_port, protocol='tcp', from_port=40003, lan_ip=None, desc=None): + if self.pmp_supported: + return self.pmp_map_port(inner_port, protocol=protocol) + + return await self.upnp_map_port( inner_port, protocol=protocol, ip=lan_ip, desc=desc) + diff --git a/appPublic/dictObject.old.py b/appPublic/dictObject.old.py new file mode 100644 index 0000000..cd7e798 --- /dev/null +++ b/appPublic/dictObject.old.py @@ -0,0 +1,161 @@ +import json +from json import JSONEncoder +from inspect import ismethod, isfunction, isbuiltin, isabstract + +def multiDict2Dict(md): + ns = {} + for k,v in md.items(): + ov = ns.get(k,None) + if ov is None: + ns[k] = v + elif type(ov) == type([]): + ov.append(v) + ns[k] = ov + else: + ns[k] = [ov,v] + return ns + +class DictObject: + def __init__(self,**kw): + self.org_keys__ = [] + self.org_keys__ = [ k for k in self.__dict__.keys()] + for k,v in kw.items(): + self.update({k:self.__DOitem(v)}) + + def __getattr__(self,name): + if name in self._addon().keys(): + return self.__getitem__(name) + return None + + def update(self,kw): + self.__dict__.update(kw) + + def _addon(self): + ks = [ k for k in self.__dict__.keys() if k not in self.org_keys__] + return {k:v for k,v in self.__dict__.items() if k in ks} + + def clear(self): + for k in self._addon().keys(): + self.__dict__.pop(k) + + def get(self,name,default=None): + return self._addon().get(name,default) + + def pop(self,k,default=None): + return self.__dict__.pop(k,default) + + def popitem(self): + return self.__dict__.popitem() + + def items(self): + return self._addon().items() + + def keys(self): + return self._addon().keys() + + def values(self): + return self._addon().values() + + def __delitem__(self,key): + self.pop(key) + + def __getitem__(self,name): + return self._addon().get(name) + + def __setitem__(self,name,value): + self.__dict__[name] = value + + def __str__(self): + return str(self._addon()) + + def __expr__(self): + return self.addon().__expr__() + + def copy(self): + return {k:v for k,v in self._addon().items()} + + def to_dict(self): + d = self._addon() + newd = self.dict_to_dict(d) + return newd + + def dict_to_dict(self,dic): + d = {} + for k,v in dic.items(): + if isinstance(v,DictObject): + d[k] = v.to_dict() + elif isinstance(v,dict): + d[k] = self.dict_to_dict(v) + elif isinstance(v,list): + d[k] = self.array_to_dict(v) + elif k == '__builtins__': + pass + elif isbuiltin(v) or isfunction(v) or ismethod(v) or isabstract(v): + pass + else: + d[k] = v + return d + + def array_to_dict(self,v): + r = [] + for i in v: + if isinstance(i,list): + r.append(self.array_to_dict(i)) + elif isinstance(i,dict): + r.append(self.dict_to_dict(i)) + elif isinstance(i,DictObject): + r.append(i.to_dict()) + elif isbuiltin(i) or isfunction(i) or ismethod(i) or isabstract(i): + pass + else: + r.append(i) + return r + + @classmethod + def isMe(self,name): + return name == 'DictObject' + + def __DOArray(self,a): + b = [ self.__DOitem(i) for i in a ] + return b + + def __DOitem(self, i): + if isinstance(i,DictObject): + return i + if isinstance(i,dict): + i = {k:v for k,v in i.items() if isinstance(k,str)} + try: + d = DictObject(**i) + return d + except Exception as e: + print("****************",i,"*******dictObject.py") + raise e + if type(i) == type([]) or type(i) == type(()) : + return self.__DOArray(i) + return i + +class DictObjectEncoder(JSONEncoder): + def default(self, o): + return o._addon() + + + +def dictObjectFactory(_klassName__,**kwargs): + def findSubclass(_klassName__,klass): + for k in klass.__subclasses__(): + if k.isMe(_klassName__): + return k + k1 = findSubclass(_klassName__,k) + if k1 is not None: + return k1 + return None + try: + if _klassName__=='DictObject': + return DictObject(**kwargs) + k = findSubclass(_klassName__,DictObject) + if k is None: + return DictObject(**kwargs) + return k(**kwargs) + except Exception as e: + print("dictObjectFactory()",e,_klassName__) + raise e