49 lines
2.0 KiB
Python
49 lines
2.0 KiB
Python
|
import logging
|
||
|
|
import os
|
||
|
|
import socket
|
||
|
|
logger = logging.getLogger()
|
||
|
|
##
|
||
|
|
import requests
|
||
|
|
from pyroute2 import IPRoute
|
||
|
|
##
|
||
|
|
from . import config
|
||
|
|
|
||
|
|
|
||
|
|
class TunnelBroker(object):
|
||
|
|
url_ip = 'https://ipv4.clientinfo.square-r00t.net/'
|
||
|
|
params_ip = {'raw': '1'}
|
||
|
|
url_api = 'https://ipv4.tunnelbroker.net/nic/update'
|
||
|
|
|
||
|
|
def __init__(self, conf_xml, tun_id = None, wan_ip = True, update = True, *args, **kwargs):
|
||
|
|
self.conf_file = os.path.abspath(os.path.expanduser(conf_xml))
|
||
|
|
logger.debug('Using config path: {0}'.format(self.conf_file))
|
||
|
|
self._conf = config.Config(self.conf_file)
|
||
|
|
if tun_id:
|
||
|
|
self.cfg = self._conf.tunnels[int(tun_id)]
|
||
|
|
else:
|
||
|
|
tun_id = list(self._conf.tunnels.keys())[0]
|
||
|
|
self.cfg = self._conf.tunnels[tun_id]
|
||
|
|
self.wan = wan_ip
|
||
|
|
self.update = update
|
||
|
|
self.my_ip = None
|
||
|
|
|
||
|
|
def _get_my_ip(self):
|
||
|
|
if self.wan:
|
||
|
|
logger.debug('WAN IP tunneling enabled; fetching WAN IP.')
|
||
|
|
req = requests.get(self.url_ip, params = self.params_ip)
|
||
|
|
if not req.ok:
|
||
|
|
logger.error('Could not fetch self IP. Request returned {0}.'.format(req.status_code))
|
||
|
|
raise RuntimeError('Could not fetch self IP')
|
||
|
|
self.my_ip = config.IP4(req.json()['ip'], 32)
|
||
|
|
logger.debug('Set my_ip to {0}.'.format(self.my_ip.str))
|
||
|
|
else:
|
||
|
|
logger.debug('WAN IP tunneling disabled; fetching LAN IP.')
|
||
|
|
ipr = IPRoute()
|
||
|
|
_defrt = ipr.get_default_routes(family = socket.AF_INET)
|
||
|
|
if len(_defrt) != 1: # This (probably) WILL fail on multipath systems.
|
||
|
|
logger.error('Could not determine default route. Does this machine have a single default route?')
|
||
|
|
raise RuntimeError('Could not determine default IPv4 route')
|
||
|
|
self.my_ip = config.IP4(_defrt[0]['attrs']['RTA_PREFSRC'], 32)
|
||
|
|
logger.debug('Set my_ip to {0}.'.format(self.my_ip.str))
|
||
|
|
return(None)
|