helpy4/bombsquad-plugin-manager
1
0
Fork 0
mirror of https://github.com/bombsquad-community/plugin-manager.git synced 2025-10-08 14:54:36 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Update natpmp_upnp.py

Browse source
This commit is contained in:
brostos 2024-03-03 20:55:34 +03:00 • committed by GitHub
parent f469c3d302
commit 2b70c7e906
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 40 additions and 25 deletions
Download patch file Download diff file Expand all files Collapse all files

65
plugins/utilities/natpmp_upnp.py
View file

@ -156,9 +156,9 @@ def add_port_mapping():
try: try:
natpmp.map_port( natpmp.map_port(
natpmp.NATPMP_PROTOCOL_UDP, natpmp.NATPMP_PROTOCOL_UDP,
43210, BS_PORT,
43210, BS_PORT,
0, 14400,
gateway_ip=natpmp.get_gateway_addr(), gateway_ip=natpmp.get_gateway_addr(),
) )
if confirm_port(): if confirm_port():
@ -168,17 +168,18 @@ def add_port_mapping():
except (NATPMPUnsupportedError, NATPMPNetworkError): except (NATPMPUnsupportedError, NATPMPNetworkError):
import upnpy import upnpy
from upnpy.exceptions import SOAPError from upnpy.exceptions import SOAPError
from urllib.error import HTTPError from urllib.error import HTTPError
upnp = upnpy.UPnP() upnp = upnpy.UPnP()
devices = upnp.discover() devices = upnp.discover()
if devices == []: if devices == []:
babase.screenmessage( babase.screenmessage(
"Please enable upnp service on your router", (1.00, 0.15, 0.15) "Please enable upnp service on your router", (1.00, 0.15, 0.15)
) )
# bui.getsound('shieldDown').play() -> RuntimeError : Sound creation failed # bui.getsound('shieldDown').play() -> RuntimeError : Sound creation failed
return return
local_ip = ( local_ip = (
( (
@ -216,7 +217,8 @@ def add_port_mapping():
) )
return return
except SOAPError: except SOAPError:
pass if confirm_port():
return
service.AddPortMapping( service.AddPortMapping(
NewRemoteHost="", NewRemoteHost="",
NewExternalPort=BS_PORT, NewExternalPort=BS_PORT,
@ -225,7 +227,7 @@ def add_port_mapping():
NewInternalClient=str(local_ip), NewInternalClient=str(local_ip),
NewEnabled="1", NewEnabled="1",
NewPortMappingDescription="Bombsquad", NewPortMappingDescription="Bombsquad",
NewLeaseDuration=0, NewLeaseDuration=14400,
) )
if confirm_port(): if confirm_port():
babase.screenmessage( babase.screenmessage(
@ -233,29 +235,42 @@ def add_port_mapping():
) )
bui.getsound("shieldUp").play() bui.getsound("shieldUp").play()
except (SOAPError, HTTPError, UnicodeDecodeError): except (SOAPError, HTTPError, UnicodeDecodeError):
babase.screenmessage('You will need to manualy add the port on the router :(') babase.screenmessage('You will need to manualy add the port at the router :(')
@threaded @threaded
def delete_port_mapping(): def delete_port_mapping():
import upnpy import socket
from upnpy.exceptions import SOAPError import natpmp
from natpmp import NATPMPUnsupportedError, NATPMPNetworkError
upnp = upnpy.UPnP()
devices = upnp.discover()
if devices == []:
return
try: try:
for upnp_dev in devices: natpmp.map_port(
for service in upnp_dev.services: natpmp.NATPMP_PROTOCOL_UDP,
if service in WAN_SERVICE_NAMES: BS_PORT,
service = upnp_dev[service] BS_PORT,
service.DeletePortMapping( 0,
NewRemoteHost="", NewExternalPort=BS_PORT, NewProtocol="UDP") gateway_ip=natpmp.get_gateway_addr(),
except: )
pass except (NATPMPUnsupportedError, NATPMPNetworkError):
import upnpy
from upnpy.exceptions import SOAPError
upnp = upnpy.UPnP()
devices = upnp.discover()
if devices == []:
return
try:
for upnp_dev in devices:
for service in upnp_dev.services:
if service in WAN_SERVICE_NAMES:
service = upnp_dev[service]
service.DeletePortMapping(NewRemoteHost="", NewExternalPort=BS_PORT, NewProtocol="UDP")
except:
pass
# ba_meta export babase.Plugin # ba_meta export babase.Plugin
@ -266,7 +281,7 @@ class Joinable(babase.Plugin):
return return
else: else:
add_port_mapping() add_port_mapping()
def on_app_shutdown(self) -> None: def on_app_shutdown(self) -> None:
delete_port_mapping() delete_port_mapping()