285 lines
8.3 KiB
Python
285 lines
8.3 KiB
Python
import collections
|
|
import copy
|
|
import ipaddress
|
|
import os
|
|
import re
|
|
##
|
|
import netaddr
|
|
import requests
|
|
from lxml import etree
|
|
from pyroute2 import IPRoute
|
|
|
|
|
|
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.iface_idx = None
|
|
self.parse()
|
|
|
|
def _iface(self):
|
|
_iface_txt = self.xml.attrib['iface']
|
|
self.iface = _iface_txt.strip()
|
|
ipr = IPRoute()
|
|
self.iface_idx = ipr.link_lookup(ifname = self.iface)[0]
|
|
ipr.close()
|
|
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 = collections.OrderedDict()
|
|
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)
|