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() # TODO: do I instead want to use HEConfig() and fetch the single unified config? # Pros: # * I wouldn't completely die on a misconfigured tunnel in the user config. # Cons: # * We'd have to skip missing tunnels (bad auth at HE, etc.) # * We would use more memory and take more time during init. 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 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)