routerbox/utils/he_ipv6/ra.py

165 lines
5.6 KiB
Python
Raw Normal View History

2020-05-15 18:01:03 -04:00
import logging
import os
import subprocess
import warnings
##
import jinja2
logger = logging.getLogger()
def_tpl_dir = os.path.join(os.path.dirname(os.path.abspath(os.path.expanduser(__file__))), 'tpl')
class RA(object):
def __init__(self, conf = None, tpl_name = None, tpl_dir = None, *args, **kwargs):
self.conf = RAConf(conf = conf, tpl_name = tpl_name, tpl_dir = tpl_dir, *args, **kwargs)
self.svc = RASvc()
class RAConf(object):
def_tpl_name = None
def_conf = None
cfgstr = None
def __init__(self, conf = None, tpl_name = None, tpl_dir = None, *args, **kwargs):
for k in ('name', 'dir'):
n = 'tpl_{0}'.format(k)
d = 'def_tpl_{0}'.format(k)
v = locals()[k]
if not v:
setattr(self, n, getattr(self, d))
else:
setattr(self, n, v)
if not conf:
self.conf = self.def_conf
else:
self.conf = os.path.abspath(os.path.expanduser(conf))
def ext_init(self):
self.tpl_dir = os.path.abspath(os.path.expanduser(self.tpl_dir))
self.loader = jinja2.FileSystemLoader(self.tpl_dir)
self.tpl_env = jinja2.Environment(loader = self.loader)
self.tpl = self.tpl_env.get_template(self.tpl_name)
return(None)
def generate(self, assignments):
ns = {}
for a in assignments:
if len(a.iface_addrs) > 3:
ns_addrs = a.iface_addrs[:3]
else:
ns_addrs = a.iface_addrs
ns[a.iface] = ns_addrs
self.cfgstr = self.tpl.render(assignments = assignments, nameservers = ns)
def write(self):
if not self.cfgstr:
raise RuntimeError('Must run .generate() first')
os.makedirs(os.path.dirname(self.conf), exist_ok = True, mode = 0o0700)
with open(self.conf, 'w') as fh:
fh.write(self.cfgstr)
class RASvc(object):
name = None
is_systemd = False
cmd_tpl = None
has_pkill = False
start_cmd = None
stop_cmd = None
restart_cmd = None
def __init__(self):
self._get_manager()
def _exec(self, cmd):
cmd_exec = subprocess.run(cmd, stdin = subprocess.PIPE, stdout = subprocess.PIPE)
if cmd_exec.returncode != 0:
logger.warning('Could not execute {0}; returned status {1}'.format(' '.join(cmd),
cmd_exec.returncode))
for i in ('stdout', 'stderr'):
s = getattr(cmd_exec, i).decode('utf-8')
if s.strip() != '':
logger.warning('{0}: {1}'.format(i.upper(), s))
return(None)
def _get_manager(self):
chkpaths = ('/run/systemd/system',
'/dev/.run/systemd',
'/dev/.systemd')
for _ in chkpaths:
if os.path.exists(_):
self.is_systemd = True
break
if self.is_systemd:
self.cmd_tpl = 'systemctl {op} {name}'
else:
# Systemd haters, if you don't understand the benefits of unified service management across all linux
# distros, you've obviously never done wide-platform management or scripting.
# Let this else block be a learning experience for you.
self.cmd_tpl = None
self.has_pkill = False
for p in os.environ.get('PATH', '/usr/bin').split(':'):
fpath = os.path.abspath(os.path.expanduser(p))
bins = os.listdir(fpath)
if 'pkill' in bins:
self.has_pkill = True
if 'service' in bins: # CentOS/RHEL pre-7.x
self.cmd_tpl = 'service {name} {op}'
break
elif 'initctl' in bins: # older Ubuntu and other Upstart distros
self.cmd_tpl = 'initctl {op} {name}'
break
elif 'rc-service' in bins: # OpenRC
self.cmd_tpl = 'rc-service {name} {op}'
break
# That wasn't even all of them.
if not self.cmd_tpl and not self.has_pkill:
logger.error('Could not find which service manager this system is using.')
raise RuntimeError('Could not determine service manager')
elif self.has_pkill: # Last-ditch effort.
self.start_cmd = [self.name]
self.stop_cmd = ['pkill', self.name]
self.restart_cmd = ['pkill', '-HUP', self.name]
else:
for k in ('start', 'stop', 'restart'):
setattr(self,
'{0}_cmd'.format(k),
self.cmd_tpl.format(name = self.name, op = k).split())
return(None)
def restart(self):
cmd = self.restart_cmd
self._exec(cmd)
return(None)
def start(self):
cmd = self.start_cmd
self._exec(cmd)
return(None)
def stop(self):
cmd = self.stop_cmd
self._exec(cmd)
return(None)
class RADVD(RA):
name = 'radvd'
def_conf = '/etc/radvd.conf'
def_tpl_name = 'radvd.conf.j2'
def __init__(self, conf = None, tpl_name = None, tpl_dir = None):
super().__init__(conf = conf, tpl_name = tpl_name, tpl_dir = tpl_dir)
self.conf.ext_init()
class DNSMasq(RA):
name = 'dnsmasq'
def_conf = '/etc/dnsmasq.d/ra.conf'
def_tpl_name = 'dnsmasq.include.j2'
def __init__(self, conf = None, tpl_name = None, tpl_dir = None):
super().__init__(conf = conf, tpl_name = tpl_name, tpl_dir = tpl_dir)
self.conf.ext_init()