Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
console = True if debug else False
level = logging.DEBUG if debug else logging.INFO
logger.init(level, console, join(self.platform_config.get_platform_log()))
self.user_platform_config = PlatformUserConfig()
self.log_aggregator = Aggregator(self.platform_config)
self.platform_app_paths = AppPaths(PLATFORM_APP_NAME, self.platform_config)
self.platform_app_paths.get_data_dir()
self.versions = Versions(self.platform_config)
self.redirect_service = RedirectService(self.user_platform_config, self.versions)
self.port_config = PortConfig(self.platform_app_paths.get_data_dir())
self.nat_pmp_port_mapper = NatPmpPortMapper()
self.upnp_port_mapper = UpnpPortMapper(UPnP())
self.port_mapper_factory = PortMapperFactory(self.nat_pmp_port_mapper, self.upnp_port_mapper)
self.port_drill_factory = PortDrillFactory(self.user_platform_config, self.port_config,
self.port_mapper_factory)
self.device_info = DeviceInfo(self.user_platform_config, self.port_config)
self.snap = Snap(self.platform_config, self.device_info)
self.platform_cron = PlatformCron(self.platform_config)
self.systemctl = Systemctl(self.platform_config)
self.ldap_auth = LdapAuth(self.platform_config, self.systemctl)
self.event_trigger = EventTrigger(self.snap, self.platform_config)
self.nginx = Nginx(self.platform_config, self.systemctl, self.device_info)
self.certbot_genetator = CertbotGenerator(self.platform_config, self.user_platform_config,
self.device_info, self.snap)
self.tls = CertificateGenerator(self.platform_config, self.user_platform_config, self.device_info, self.nginx,
self.certbot_genetator)
self.device = Device(self.platform_config, self.user_platform_config, self.redirect_service,
def action_remove_forward(arg):
if arg == True:
global debug
debug = True
# Check if UPNP is available
upnpc = miniupnpc.UPnP()
# Discover UPNP devices
upnpc.discoverdelay = 200
if debug:
print 'Check UPNP Support'
pub_ip = None
try:
ndevices = upnpc.discover()
if debug:
print ' %d UPNP device(s) detected' % (ndevices)
if ndevices:
upnpc.selectigd()
pub_ip = upnpc.externalipaddress()
def upnp_start(self):
try:
u = miniupnpc.UPnP()
u.discoverdelay = 200
logging.info("Scanning for UPnP devices")
if u.discover() > 0:
logging.info("Device discovered")
u.selectigd()
if self.ipv6 == "" and self.ip != struct.unpack("!I", socket.inet_aton(u.externalipaddress()))[0]:
logging.warning(
"Mismatched external address, more than one layers of NAT? UPnP may not work.")
self.upnp_mapping(u)
else:
logging.error("No UPnP devices discovered")
except Exception:
logging.error("Error arose in UPnP discovery")
def get_external_ip_and_setup_upnp():
try:
u = miniupnpc.UPnP()
u.discoverdelay = 200
u.discover()
u.selectigd()
if u.getspecificportmapping(4444, "UDP"):
u.deleteportmapping(4444, "UDP")
log.info("Removed UPnP redirect for UDP 4444.")
u.addportmapping(4444, 'UDP', u.lanaddr, 4444, 'LBRY DHT port', '')
log.info("got external ip from upnp")
return u.externalipaddress()
except Exception:
log.exception("derp")
r = requests.get('https://api.ipify.org', {'format': 'json'})
log.info("got external ip from ipify.org")
return r.json()['ip']
def _upnp_igd_connect():
logger.debug("TrSv", "Attempting to connect to uPnP IGD")
upnpc = miniupnpc.UPnP()
upnpc.discoverdelay = 3000
devs = upnpc.discover()
if devs == 0:
msg = "Failed to connect to uPnP IGD: no devices found"
logger.warning("TrSv", msg)
return
try:
upnpc.selectigd()
except Exception as e:
msg = "Failed to connect to uPnP IGD: {0}"
logger.warning("TrSv", msg.format(str(e)))
return upnpc
def AddPortMappingModule(self):
"""Function to create a Port Mapping via the python binding: miniupnpc.
IGDv1: If a Port Mapping already exist:
It's updated with a new static port mapping that does not expire.
IGDv2: If a Port Mapping already exist:
It's updated with a new lease duration of 7 days.
"""
import miniupnpc
u = miniupnpc.UPnP()
u.discoverdelay = self.discoverdelay
# Discovering devices
log.adddebug('Discovering... delay=%sms' % u.discoverdelay)
try:
log.adddebug('%s device(s) detected' % u.discover())
except Exception as e:
raise RuntimeError(
_('UPnP exception (should never happen): %(error)s') %
{'error': str(e)})
# Select an IGD
try:
u.selectigd()
except Exception as e:
def upnp_add(port):
'''
:param port: local port
:return: `None` if failed, `external_ip, external_port` if succeed
'''
log_net.debug('Setting UPNP')
import miniupnpc
upnpc = miniupnpc.UPnP()
upnpc.discoverdelay = 200
ndevices = upnpc.discover()
log_net.debug('UPNP device(s) detected', num=ndevices)
if not ndevices:
return None
upnpc.selectigd()
external_ip = upnpc.externalipaddress()
log_net.debug('upnp', external_ip=external_ip,
status=upnpc.statusinfo(),
connection_type=upnpc.connectiontype())
# find a free port for the redirection
external_port = port
found = False