import copy import ipaddress import os import re ## import netaddr import requests from lxml import etree 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() class Allocation(object): def __init__(self, alloc_xml): self.xml = alloc_xml self.prefix = None self.ip = None self.iface = None self.parse() def _iface(self): _iface_txt = self.xml.attrib['iface'] self.iface = _iface_txt.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._iface() self._ip() return(None) class Tunnel(object): def __init__(self, tun_xml): 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.allocations = [] self.parse() def _allocations(self): for _allocation_xml in self.xml.findall('allocations'): self.allocations.append(Allocation(_allocation_xml)) 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 _id(self): self.id = int(self.xml.attrib['id'].strip()) 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._client() self._server() self._allocations() return(None) 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 Config(object): 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') self.tree = None self.ns_tree = None self.xml = None self.ns_xml = None self.raw = None self.xsd = None self.defaults_parser = None self.obj = None self.tunnels = {} self.creds = {} self.parse() 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 = Tunnel(tun_xml) tun.creds = self.creds.get(tun.creds) self.tunnels[tun.id] = tun return(None) 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(self): self.parse_raw() self.get_xsd() self.populate_defaults() self.validate() self.subparse() return(None) def parse_raw(self, parser = None): if not self.raw: with open(self.xml_path, 'rb') as fh: self.raw = fh.read() 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 subparse(self): self._creds() self._tunnels() return(None) def validate(self): if not self.xsd: self.get_xsd() self.xsd.assertValid(self.ns_tree) return(None)