import collections import copy import ipaddress import os import re ## import netaddr import requests import requests.auth from lxml import etree from pyroute2 import IPRoute ## from . import tunnel from . import radvd from . import utils 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): raise ValueError('xml_path does not exist') 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. self.tun_xml = self.xml.find('tunnel') # Will only return a single for this URL. 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('clientv4').text if _client is not None and _client.strip() != '': self.client = tunnel.IP4(_client.strip(), 32) 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'), 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)