386 lines
14 KiB
Python
386 lines
14 KiB
Python
import collections
|
|
import copy
|
|
import os
|
|
import re
|
|
##
|
|
import requests
|
|
import requests.auth
|
|
from lxml import etree
|
|
##
|
|
from . import tunnel
|
|
|
|
|
|
def create_default_cfg():
|
|
# Create a stripped sample config.
|
|
ws_re = re.compile(r'^\s*$')
|
|
cur_dir = os.path.dirname(os.path.abspath(os.path.expanduser(__file__)))
|
|
xtbxml = os.path.join(cur_dir, 'example.tunnelbroker.xml')
|
|
with open(xtbxml, 'rb') as fh:
|
|
xml = etree.fromstring(fh.read())
|
|
# Create a stripped sample config.
|
|
# First we strip comments (and fix the ensuing whitespace).
|
|
# etree has a .canonicalize(), but it chokes on a default namespace.
|
|
# https://bugs.launchpad.net/lxml/+bug/1869455
|
|
# So everything we do is kind of a hack.
|
|
# for c in xml.xpath("//comment()"):
|
|
# parent = c.getparent()
|
|
# parent.remove(c)
|
|
xmlstr = etree.tostring(xml, with_comments = False, method = 'c14n', pretty_print = True).decode('utf-8')
|
|
newstr = []
|
|
for line in xmlstr.splitlines():
|
|
r = ws_re.search(line)
|
|
if not r:
|
|
newstr.append(line.strip())
|
|
xml = etree.fromstring(''.join(newstr).encode('utf-8'))
|
|
# Remove text and attr text.
|
|
xpathq = "descendant-or-self::*[namespace-uri()!='']"
|
|
for e in xml.xpath(xpathq):
|
|
if e.tag == '{{{0}}}heIPv6'.format(xml.nsmap[None]):
|
|
continue
|
|
if e.text is not None and e.text.strip() != '':
|
|
e.text = ''
|
|
for k, v in e.attrib.items():
|
|
if v is not None:
|
|
e.attrib[k] = ''
|
|
# Remove multiple children of same type to simplify.
|
|
for e in xml.xpath(xpathq):
|
|
if e.tag == '{{{0}}}heIPv6'.format(xml.nsmap[None]):
|
|
continue
|
|
parent = e.getparent()
|
|
try:
|
|
for idx, child in enumerate(parent.findall(e.tag)):
|
|
if idx == 0:
|
|
continue
|
|
parent.remove(child)
|
|
except AttributeError:
|
|
pass
|
|
# And add a comment pointing them to the fully commented config.
|
|
xml.insert(0, etree.Comment(('\n Please reference the fully commented example.tunnelbroker.xml found either '
|
|
'at:\n '
|
|
' * {0}\n * https://git.square-r00t.net/RouterBox/tree/utils/he_ipv6/'
|
|
'example.tunnelbroker.xml\n and then configure this according to those '
|
|
'instructions.\n ').format(xtbxml)))
|
|
return(etree.tostring(xml,
|
|
pretty_print = True,
|
|
with_comments = True,
|
|
with_tail = True,
|
|
encoding = 'UTF-8',
|
|
xml_declaration = True))
|
|
|
|
|
|
class Credential(object):
|
|
def __init__(self, cred_xml):
|
|
self.xml = cred_xml
|
|
self.id = None
|
|
self.user = None
|
|
self.password = None
|
|
self.parse()
|
|
|
|
def _id(self):
|
|
_id = self.xml.attrib.get('id')
|
|
if not _id:
|
|
raise ValueError('Missing required id attribute')
|
|
self.id = _id.strip()
|
|
return(None)
|
|
|
|
def _password(self):
|
|
_key_xml = self.xml.find('password')
|
|
if _key_xml is None:
|
|
raise ValueError('Missing required password element')
|
|
_key_txt = _key_xml.text
|
|
if not _key_txt:
|
|
raise ValueError('password element is empty')
|
|
self.password = _key_txt.strip()
|
|
return(None)
|
|
|
|
def _user(self):
|
|
_user_xml = self.xml.find('user')
|
|
if _user_xml is None:
|
|
raise ValueError('Missing required user element')
|
|
_user_txt = _user_xml.text
|
|
if not _user_txt:
|
|
raise ValueError('user element is empty')
|
|
self.user = _user_txt.strip()
|
|
return(None)
|
|
|
|
def parse(self):
|
|
self._id()
|
|
self._user()
|
|
self._password()
|
|
return(None)
|
|
|
|
|
|
class BaseConfig(object):
|
|
default_xsd = None
|
|
|
|
def __init__(self, xml_raw, *args, **kwargs):
|
|
self.raw = xml_raw
|
|
self.tree = None
|
|
self.ns_tree = None
|
|
self.xml = None
|
|
self.ns_xml = None
|
|
self.xsd = None
|
|
self.defaults_parser = None
|
|
self.obj = None
|
|
self.parse_xml()
|
|
|
|
def get_xsd(self):
|
|
raw_xsd = None
|
|
base_url = None
|
|
xsi = self.xml.nsmap.get('xsi', 'http://www.w3.org/2001/XMLSchema-instance')
|
|
schemaLocation = '{{{0}}}schemaLocation'.format(xsi)
|
|
schemaURL = self.xml.attrib.get(schemaLocation, self.default_xsd)
|
|
split_url = schemaURL.split()
|
|
if len(split_url) == 2: # a properly defined schemaLocation
|
|
schemaURL = split_url[1]
|
|
else:
|
|
schemaURL = split_url[0] # a LAZY schemaLocation
|
|
if schemaURL.startswith('file://'):
|
|
schemaURL = re.sub(r'^file://', r'', schemaURL)
|
|
with open(schemaURL, 'rb') as fh:
|
|
raw_xsd = fh.read()
|
|
base_url = os.path.dirname(schemaURL) + '/'
|
|
else:
|
|
req = requests.get(schemaURL)
|
|
if not req.ok:
|
|
raise RuntimeError('Could not download XSD')
|
|
raw_xsd = req.content
|
|
base_url = os.path.split(req.url)[0] + '/' # This makes me feel dirty.
|
|
self.xsd = etree.XMLSchema(etree.XML(raw_xsd, base_url = base_url))
|
|
return(None)
|
|
|
|
def parse_xml(self):
|
|
self.parse_raw()
|
|
self.get_xsd()
|
|
self.populate_defaults()
|
|
self.validate()
|
|
return(None)
|
|
|
|
def parse_raw(self, parser = None):
|
|
self.xml = etree.fromstring(self.raw, parser = parser)
|
|
self.ns_xml = etree.fromstring(self.raw, parser = parser)
|
|
self.tree = self.xml.getroottree()
|
|
self.ns_tree = self.ns_xml.getroottree()
|
|
self.tree.xinclude()
|
|
self.ns_tree.xinclude()
|
|
self.strip_ns()
|
|
return(None)
|
|
|
|
def populate_defaults(self):
|
|
if not self.xsd:
|
|
self.get_xsd()
|
|
if not self.defaults_parser:
|
|
self.defaults_parser = etree.XMLParser(schema = self.xsd, attribute_defaults = True)
|
|
self.parse_raw(parser = self.defaults_parser)
|
|
return(None)
|
|
|
|
def remove_defaults(self):
|
|
self.parse_raw()
|
|
return(None)
|
|
|
|
def strip_ns(self, obj = None):
|
|
# https://stackoverflow.com/questions/30232031/how-can-i-strip-namespaces-out-of-an-lxml-tree/30233635#30233635
|
|
xpathq = "descendant-or-self::*[namespace-uri()!='']"
|
|
if not obj:
|
|
for x in (self.tree, self.xml):
|
|
for e in x.xpath(xpathq):
|
|
e.tag = etree.QName(e).localname
|
|
elif isinstance(obj, (etree._Element, etree._ElementTree)):
|
|
obj = copy.deepcopy(obj)
|
|
for e in obj.xpath(xpathq):
|
|
e.tag = etree.QName(e).localname
|
|
return(obj)
|
|
else:
|
|
raise ValueError('Did not know how to parse obj parameter')
|
|
return(None)
|
|
|
|
def validate(self):
|
|
if not self.xsd:
|
|
self.get_xsd()
|
|
self.xsd.assertValid(self.ns_tree)
|
|
return(None)
|
|
|
|
|
|
class Config(BaseConfig):
|
|
default_xsd = 'http://schema.xml.r00t2.io/projects/he_ipv6.xsd'
|
|
|
|
def __init__(self, xml_path, *args, **kwargs):
|
|
self.xml_path = os.path.abspath(os.path.expanduser(xml_path))
|
|
if not os.path.isfile(self.xml_path):
|
|
with open(self.xml_path, 'wb') as fh:
|
|
fh.write(create_default_cfg())
|
|
raise ValueError('xml_path does not exist; '
|
|
'a sample configuration has been generated (be sure to configure it)')
|
|
else:
|
|
with open(xml_path, 'rb') as fh:
|
|
raw_xml = fh.read()
|
|
super().__init__(raw_xml, *args, **kwargs)
|
|
self.creds = {}
|
|
self.tunnels = collections.OrderedDict()
|
|
self.subparse()
|
|
|
|
def _creds(self):
|
|
creds_xml = self.xml.find('creds')
|
|
for cred_xml in creds_xml.findall('cred'):
|
|
cred = Credential(cred_xml)
|
|
self.creds[cred.id] = cred
|
|
return(None)
|
|
|
|
def _tunnels(self):
|
|
tunnels_xml = self.xml.find('tunnels')
|
|
for tun_xml in tunnels_xml.findall('tunnel'):
|
|
tun_id = int(tun_xml.attrib['id'].strip())
|
|
tun_creds_id = tun_xml.attrib['creds']
|
|
creds = self.creds[tun_creds_id]
|
|
update_key = tun_xml.find('updateKey').text.strip()
|
|
he_conf = HETunnelConfig(tun_id, creds, update_key)
|
|
tun = tunnel.Tunnel(tun_xml, he_conf, self.creds[tun_creds_id])
|
|
self.tunnels[tun_id] = tun
|
|
return(None)
|
|
|
|
def subparse(self):
|
|
self._creds()
|
|
self._tunnels()
|
|
return(None)
|
|
|
|
|
|
class HEBaseConfig(BaseConfig):
|
|
default_xsd = 'http://schema.xml.r00t2.io/projects/tunnelbroker.xsd'
|
|
nsmap = {None: 'https://tunnelbroker.net/tunnelInfo.php',
|
|
'xsi': 'http://www.w3.org/2001/XMLSchema-instance'}
|
|
attr_qname = etree.QName('http://www.w3.org/2001/XMLSchema-instance', 'schemaLocation')
|
|
schema_loc = 'https://tunnelbroker.net/tunnelInfo.php {0}'.format(default_xsd)
|
|
url = ''
|
|
|
|
def __init__(self, creds, *args, **kwargs):
|
|
self.creds = creds
|
|
super().__init__(self._fetch(), *args, **kwargs)
|
|
|
|
def _add_ns(self, raw_xml):
|
|
# https://mailman-mail5.webfaction.com/pipermail/lxml/20100323/013260.html
|
|
_xml = etree.fromstring(raw_xml)
|
|
_nsmap = copy.deepcopy(_xml.nsmap)
|
|
_nsmap.update(self.nsmap)
|
|
mod_xml = etree.Element(_xml.tag,
|
|
{self.attr_qname: self.schema_loc},
|
|
nsmap = _nsmap)
|
|
mod_xml[:] = _xml[:]
|
|
return(etree.tostring(mod_xml,
|
|
encoding = 'UTF-8',
|
|
xml_declaration = True,
|
|
pretty_print = True,
|
|
with_tail = True,
|
|
with_comments = True))
|
|
|
|
def _fetch(self):
|
|
req = requests.get(self.url,
|
|
auth = requests.auth.HTTPBasicAuth(self.creds.user,
|
|
self.creds.password))
|
|
if not req.ok:
|
|
raise RuntimeError('Could not fetch remote tunnel information')
|
|
raw_xml = self._add_ns(req.content)
|
|
return(raw_xml)
|
|
|
|
|
|
# This isn't really used.
|
|
class HEConfig(HEBaseConfig):
|
|
default_xsd = 'http://schema.xml.r00t2.io/projects/tunnelbroker.xsd'
|
|
nsmap = {None: 'https://tunnelbroker.net/tunnelInfo.php',
|
|
'xsi': 'http://www.w3.org/2001/XMLSchema-instance'}
|
|
attr_qname = etree.QName('http://www.w3.org/2001/XMLSchema-instance', 'schemaLocation')
|
|
schema_loc = 'https://tunnelbroker.net/tunnelInfo.php {0}'.format(default_xsd)
|
|
url = 'https://tunnelbroker.net/tunnelInfo.php'
|
|
|
|
def __init__(self, creds, *args, **kwargs):
|
|
super().__init__(creds, *args, **kwargs)
|
|
self.tunnels = {}
|
|
|
|
def add_tunnel(self, tun_id, update_key):
|
|
self.tunnels[tun_id] = HETunnelConfig(tun_id, self.creds, update_key)
|
|
return(None)
|
|
|
|
|
|
class HETunnelConfig(HEBaseConfig):
|
|
# default_xsd = 'http://schema.xml.r00t2.io/projects/tunnelbroker.tun.xsd'
|
|
# nsmap = {None: 'https://tunnelbroker.net/tunnelInfo.php?tid',
|
|
# 'xsi': 'http://www.w3.org/2001/XMLSchema-instance'}
|
|
# attr_qname = etree.QName('http://www.w3.org/2001/XMLSchema-instance', 'schemaLocation')
|
|
# schema_loc = 'https://tunnelbroker.net/tunnelInfo.php?tid {0}'.format(default_xsd)
|
|
url = 'https://tunnelbroker.net/tunnelInfo.php?tid={0}'
|
|
|
|
def __init__(self, tun_id, creds, update_key, *args, **kwargs):
|
|
self.tun_id = int(tun_id)
|
|
self.url = self.url.format(self.tun_id)
|
|
self.creds = copy.deepcopy(creds)
|
|
self.creds.password = update_key
|
|
super().__init__(self.creds, *args, **kwargs)
|
|
self.id = None
|
|
self.description = None
|
|
self.client = None # Client IPv6
|
|
self.server = None # Server IPv6
|
|
self.endpoint = None # Server IPv4
|
|
self.my_ip = None # Client IPv4 (not necessary; we locally cache Tunnel.my_ip)
|
|
self.allocations = {} # keys are 64 and 48
|
|
self.rdns = [] # Also not necessary, but it's in the XML so why not.
|
|
# Will only return a single <tunnel> for this URL.
|
|
# TODO: I can probably consolidate all this into HECond instead?
|
|
self.tun_xml = self.xml.find('tunnel')
|
|
self.parse()
|
|
|
|
def _alloc(self):
|
|
for a in ('64', '48'):
|
|
_alloc = self.tun_xml.find('routed{0}'.format(a))
|
|
if _alloc is not None and _alloc.text.strip() != '':
|
|
self.allocations[int(a)] = tunnel.Allocation(_alloc.text.strip())
|
|
return(None)
|
|
|
|
def _client(self):
|
|
_client = self.tun_xml.find('clientv6').text
|
|
if _client is not None and _client.strip() != '':
|
|
self.client = tunnel.IP6(_client.strip(), 64)
|
|
return(None)
|
|
|
|
def _desc(self):
|
|
_desc = self.tun_xml.find('description').text
|
|
if _desc is not None and _desc.strip() != '':
|
|
self.description = _desc.strip()
|
|
return(None)
|
|
|
|
def _endpoint(self):
|
|
self.endpoint = tunnel.IP4(self.tun_xml.find('serverv4').text.strip(), 32)
|
|
return(None)
|
|
|
|
def _id(self):
|
|
self.id = int(self.tun_xml.attrib['id'])
|
|
return(None)
|
|
|
|
def _my_ip(self):
|
|
_ip = self.tun_xml.find('clientv4').text
|
|
if _ip is not None and _ip.strip() != '':
|
|
self.my_ip = tunnel.IP4(_ip.strip(), 32)
|
|
return(None)
|
|
|
|
def _rdns(self):
|
|
self.rdns = []
|
|
for r in range(1, 6):
|
|
_rdns = self.tun_xml.find('rdns{0}'.format(r))
|
|
if _rdns is not None and _rdns.text.strip() != '':
|
|
self.rdns.append(_rdns.text.strip())
|
|
self.rdns = tuple(self.rdns)
|
|
return(None)
|
|
|
|
def _server(self):
|
|
self.server = tunnel.IP6(self.tun_xml.find('serverv6').text.strip(), 128)
|
|
return(None)
|
|
|
|
def parse(self):
|
|
self._id()
|
|
self._client()
|
|
self._desc()
|
|
self._endpoint()
|
|
self._server()
|
|
self._my_ip()
|
|
self._alloc()
|
|
self._rdns()
|
|
return(None)
|