routerbox/utils/he_ipv6/config.py

import collections
import copy
import ipaddress
import os
import re
##
import netaddr
import requests
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 == ('t', '1'):
        return(True)
    elif xml_str == ('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 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.alloc = None  # This must be set externally to a mapped Allocation instance
        self.alloc_name = None
        self.prefix = None
        self.iface_ip = None
        self.alloc_block = None
        self.parse()

    def _alloc(self):
        self.alloc_name = self.xml.attrib['alloc'].strip()
        return(None)

    def _iface(self):
        self.iface = self.xml.attrib['iface'].strip()
        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.iface_ip = IP6(str(next(self.alloc.ip.net.hosts())), 128)
        self.alloc_block = self.alloc.ip.alloc_block
        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 _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._id()
        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.creds_id = None
        self.allocations = {}
        self.assignments = []
        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')
        radvd = xml2bool(_assigns_xml.attrib.get('radvd', 'false'))
        radvd_dns = xml2bool(_assigns_xml.attrib.get('radvdDns', 'false'))
        for _assign_xml in _assigns_xml.findall('assign'):
            assign = Assignment(_assign_xml, radvd = radvd, dns = 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 _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()
        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_id)
            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)