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 radvd def xml2bool(xml_str): if xml_str is None: return(None) xml_str = xml_str.lower()[0] if xml_str in ('t', '1'): return(True) elif xml_str in ('f', '0'): return(False) else: raise ValueError('Not a boolean value') class IP(object): type = None version = None _ip = ipaddress.ip_address _net = ipaddress.ip_network _net_ip = netaddr.IPAddress _net_net = netaddr.IPNetwork def __init__(self, ip, prefix, *args, **kwargs): self.str = ip self.prefix = int(prefix) self.net_ip = self._net_ip(self.str) self.net_net = self._net_net('{0}/{1}'.format(self.str, self.prefix)) def _ext_init(self): self.ip = self._ip(self.str) self.net = self._net('{0}/{1}'.format(self.str, self.prefix), strict = False) return(None) class IP4(IP): type = 'IPv4' version = 4 _ip = ipaddress.IPv4Address _net = ipaddress.IPv4Network def __init__(self, ip, prefix, *args, **kwargs): super().__init__(ip, prefix, *args, **kwargs) self._ext_init() class IP6(IP): type = 'IPv6' version = 6 _ip = ipaddress.IPv6Address _net = ipaddress.IPv6Network def __init__(self, ip, prefix, *args, **kwargs): super().__init__(ip, prefix, *args, **kwargs) self._ext_init() self.alloc_block = netaddr.SubnetSplitter(self.net_net) class Credential(object): def __init__(self, cred_xml): self.xml = cred_xml self.id = None self.user = None self.key = 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 _update_key(self): _key_xml = self.xml.find('updateKey') if _key_xml is None: raise ValueError('Missing required updateKey element') _key_txt = _key_xml.text if not _key_txt: raise ValueError('updateKey element is empty') self.key = _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._update_key() return(None) class Assignment(object): def __init__(self, assign_xml, radvd = False, dns = False): self.xml = assign_xml self.do_radvd = radvd self.radvd_dns = dns self.iface = None self.iface_idx = None self.iface_addrs = [] self.iface_blocks = [] self.alloc = None # This must be set externally to a mapped Allocation instance self.alloc_name = None self.prefix = None self.alloc_block = None self.parse() def _alloc(self): self.alloc_name = self.xml.attrib['alloc'].strip() return(None) def _iface(self): _iface_txt = self.xml.attrib['iface'].strip() self.iface = _iface_txt.strip() ipr = IPRoute() self.iface_idx = ipr.link_lookup(ifname = self.iface)[0] ipr.close() return(None) def _prefix(self): self.prefix = int(self.xml.attrib.get('prefix', 64).strip()) return(None) def parse(self): self._iface() self._alloc() self._prefix() return(None) def parse_alloc(self): self.alloc_block = self.alloc.ip.alloc_block self.iface_blocks = self.alloc_block.extract_subnet(self.prefix, count = 1) for i in self.iface_blocks: self.iface_addrs.append(IP6(str(next(i.iter_hosts())), 128)) return(None) class Allocation(object): def __init__(self, alloc_xml): self.xml = alloc_xml self.id = None self.prefix = None self.ip = None self.iface = None self.iface_idx = None self.parse() def _id(self): self.id = self.xml.attrib['id'].strip() return(None) def _ip(self): _ip_txt = self.xml.text.strip() _prefix_txt = self.xml.attrib['prefix'].strip() self.ip = IP6(_ip_txt, _prefix_txt) self.prefix = self.ip.prefix return(None) def parse(self): self._id() self._ip() return(None) class Tunnel(object): def __init__(self, tun_xml, he_config): self.xml = tun_xml self.id = None self.client = None self.server = None self.creds = None # This should be handled externally and map to a Cred obj self.creds_id = None self.radvd = None self.enable_radvd = None self.radvd_dns = None self.allocations = {} # This is a dict of {}[alloc.id] = Allocation obj self.assignments = [] # This is a list of Assignment objs self.heconf = he_config self.parse() def _allocations(self): _allocs_xml = self.xml.find('allocations') for _allocation_xml in _allocs_xml.findall('alloc'): alloc = Allocation(_allocation_xml) self.allocations[alloc.id] = alloc return(None) def _assignments(self): _assigns_xml = self.xml.find('assignments') self.enable_radvd = xml2bool(_assigns_xml.attrib.get('radvd', 'false')) self.radvd_dns = xml2bool(_assigns_xml.attrib.get('radvdDns', 'false')) for _assign_xml in _assigns_xml.findall('assign'): assign = Assignment(_assign_xml, radvd = self.enable_radvd, dns = self.radvd_dns) assign.alloc = self.allocations[assign.alloc_name] assign.parse_alloc() self.assignments.append(assign) return(None) def _client(self): _client_xml = self.xml.find('client') _ip_txt = _client_xml.text.strip() _prefix_txt = _client_xml.attrib['prefix'].strip() self.client = IP6(_ip_txt, _prefix_txt) return(None) def _creds(self): self.creds_id = self.xml.attrib['creds'].strip() return(None) def _id(self): self.id = int(self.xml.attrib['id'].strip()) return(None) def _radvd(self): self.radvd = radvd.RADVD() self.radvd.conf.generate(self.assignments) return(None) def _server(self): _server_xml = self.xml.find('server') _ip_text = _server_xml.text.strip() self.server = IP4(_ip_text, 32) return(None) def parse(self): self._id() self._creds() self._client() self._server() self._allocations() self._assignments() self._radvd() 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.heconf = None 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 _heconf(self): self.heconf = HEConfig(self.creds) 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 = Tunnel(tun_xml, self.heconf.tunnels[tun_id]) tun.creds = self.creds.get(tun.creds_id) self.tunnels[tun_id] = tun return(None) def subparse(self): self._creds() self._heconf() self._tunnels() return(None) class HEConfig(BaseConfig): default_xsd = 'http://schema.xml.r00t2.io/projects/tunnelbroker.xsd' nsmap = {None: 'https://tunelbroker.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) def __init__(self, creds, xml_url = 'https://tunnelbroker.net/tunnelInfo.php', *args, **kwargs): self.creds = creds self.url = xml_url 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) super().__init__(raw_xml, *args, **kwargs) self.tunnels = collections.OrderedDict() self.subparse() def subparse(self): pass # TODO. nativize blocks, etc. Move out of local config. 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))