routerbox/utils/he_ipv6/config.py

425 lines
13 KiB
Python
Raw Normal View History

2020-05-12 03:26:44 -04:00
import collections
2020-05-12 02:55:38 -04:00
import copy
import ipaddress
2020-05-12 01:21:52 -04:00
import os
import re
##
2020-05-12 02:55:38 -04:00
import netaddr
2020-05-12 01:21:52 -04:00
import requests
import requests.auth
2020-05-12 01:21:52 -04:00
from lxml import etree
2020-05-12 03:26:44 -04:00
from pyroute2 import IPRoute
##
from . import radvd
def xml2bool(xml_str):
if xml_str is None:
return(None)
xml_str = xml_str.lower()[0]
2020-05-13 21:10:09 -04:00
if xml_str in ('t', '1'):
return(True)
2020-05-13 21:10:09 -04:00
elif xml_str in ('f', '0'):
return(False)
else:
raise ValueError('Not a boolean value')
2020-05-12 01:21:52 -04:00
2020-05-12 02:55:38 -04:00
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)
2020-05-13 21:10:09 -04:00
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
2020-05-13 21:10:09 -04:00
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):
2020-05-13 21:10:09 -04:00
_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
2020-05-13 21:10:09 -04:00
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)
2020-05-12 02:55:38 -04:00
class Allocation(object):
def __init__(self, alloc_xml):
self.xml = alloc_xml
self.id = None
2020-05-12 02:55:38 -04:00
self.prefix = None
self.ip = None
self.iface = None
2020-05-12 03:26:44 -04:00
self.iface_idx = None
2020-05-12 02:55:38 -04:00
self.parse()
def _id(self):
self.id = self.xml.attrib['id'].strip()
return(None)
2020-05-12 02:55:38 -04:00
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()
2020-05-12 02:55:38 -04:00
self._ip()
return(None)
class Tunnel(object):
def __init__(self, tun_xml, he_config):
2020-05-12 02:55:38 -04:00
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
2020-05-12 19:39:16 -04:00
self.creds_id = None
2020-05-13 21:10:09 -04:00
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
2020-05-12 02:55:38 -04:00
self.parse()
def _allocations(self):
_allocs_xml = self.xml.find('allocations')
2020-05-12 23:17:44 -04:00
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')
2020-05-13 21:10:09 -04:00
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'):
2020-05-13 21:10:09 -04:00
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)
2020-05-12 02:55:38 -04:00
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)
2020-05-12 19:39:16 -04:00
def _creds(self):
self.creds_id = self.xml.attrib['creds'].strip()
return(None)
2020-05-12 02:55:38 -04:00
def _id(self):
self.id = int(self.xml.attrib['id'].strip())
return(None)
2020-05-13 21:10:09 -04:00
def _radvd(self):
self.radvd = radvd.RADVD()
self.radvd.conf.generate(self.assignments)
return(None)
2020-05-12 02:55:38 -04:00
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()
2020-05-12 19:39:16 -04:00
self._creds()
2020-05-12 02:55:38 -04:00
self._client()
self._server()
self._allocations()
self._assignments()
2020-05-13 21:10:09 -04:00
self._radvd()
2020-05-12 02:55:38 -04:00
return(None)
class BaseConfig(object):
default_xsd = None
2020-05-12 01:21:52 -04:00
def __init__(self, xml_raw, *args, **kwargs):
self.raw = xml_raw
2020-05-12 01:21:52 -04:00
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()
2020-05-12 01:21:52 -04:00
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):
2020-05-12 02:55:38 -04:00
self.parse_raw()
self.get_xsd()
self.populate_defaults()
self.validate()
return(None)
2020-05-12 01:21:52 -04:00
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
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))