#!/usr/bin/python3 """install.py - copy files from to_system into system directories - set host name / network settings from boxtools/cfg file - and many more """ if bytes == str: raise NotImplementedError('python2 not supported') import sys import os import filecmp import re import types import socket import tempfile import importlib from pathlib import Path from subprocess import Popen, PIPE from os.path import getmtime, exists from gitea import change_to_gitea def exit(text='please restart ./install.py again'): if text: print(text) sys.exit() try: import serial from ipaddress import IPv4Interface from utils import BoxInfo, check_service, unix_cmd, change_firewall, UndefinedConfigFile, convert_cfg except ImportError as e: pkgs = 'pyserial', 'ipaddress', 'netifaces' if 'yes'.startswith(input(f'install {", ".join(pkgs)}? [y]')): os.system('sudo apt-get -y update') os.system('sudo apt-get -y install python3-pip') for pkg in pkgs: os.system(f'sudo pip3 install --break-system-packages {pkg}') exit() else: exit('') CENTRAL_BOXTOOLS_REPO = 'l_samenv@linse-c:boxtools/' main_ports = {22} # ssh router_ports = set() TOOLS = BoxInfo.TOOLS BOX_TYPES = {'ionopi', 'ionopimax', 'control-box', 'bare-apu', 'dual-eth-rpi'} more_info = False DEL = '__to_delete__' STARTUP_TEXT = TOOLS / 'startup_display.txt' COMMENT = "; please refer to README.md for help" TEMPLATES = {} TEMPLATES['bare-apu'] = f"""{COMMENT} [BOX] type=%(typ)s MAC=%(mac)s version=1 [NETWORK] eth0=wan eth1=192.168.2.2 eth2=192.168.3.3 eth3=192.168.127.254 [ROUTER] 3001=192.168.127.254:3001 """ TEMPLATES['control-box'] = f"""{COMMENT} [BOX] type=%(typ)s MAC=%(mac)s version=1 [NETWORK] eth0=wan eth1=192.168.1.1 eth2=192.168.2.2 eth3=192.168.3.3 [DISPLAY] line0=startup... line1=HOST line2=ADDR """ TEMPLATES['dual-eth-rpi'] = f"""{COMMENT} [BOX] type=%(typ)s MAC=%(mac)s version=1 [NETWORK] end0=wan %(mac2)s=192.168.1.1 """ GENERIC_TEMPLATE = f"""{COMMENT} [BOX] type=%(typ)s MAC=%(mac)s version=1 """ DHCP_HEADER = """ default-lease-time 600; max-lease-time 7200; authoritative; """ class Service: """base class for services managed by this script - name: name of the systemd unit (default: lower case class name) - the section in the cfg file is the upper case class name - __init__ gets the section content as keyword arguments, registers the needed ports/packages, and sets servicecfg = None when the service is not wanted - activate: when False, the service is only enabled, but not started (or restarted) by this script, i.e. it runs after the next boot (or when another unit pulls it in) """ name = None as_root = False activate = True servicecfg = None def __init_subclass__(cls, **kwds): super().__init_subclass__(**kwds) if 'name' not in cls.__dict__: cls.name = cls.__name__.lower() def handle(self, to_start, reload_systemd): service = self.name active, enabled = check_service(service, as_root=self.as_root) if self.servicecfg is None: if active or enabled: check_service(service, False, doit, as_root=self.as_root) show.dirty = True return if not enabled: if not doit: print(service, 'is not enabled') to_start[service] = 'enable', self.as_root elif not active and self.activate: if not doit: print(service, 'is not active') to_start[service] = 'restart', self.as_root if self.as_root: systemdir = Path('/etc/systemd/system') else: systemdir = Path('~/.config/systemd/user').expanduser() if write_when_new(systemdir / f'{service}.service', self.servicecfg, as_root=self.as_root): show.dirty = True reload_systemd.add(self.as_root) if self.activate: # a changed or new unit file needs a (re)start. without activate, it is # enough to have the unit enabled (see above) and systemd reloaded to_start[service] = 'restart', self.as_root class Router(Service): as_root = True def __init__(self, **opts): if not opts: self.servicecfg = None return for key in opts: try: router_ports.add(int(key)) except ValueError: pass router_ports.update((1110, 1111, 1112)) servicecfg = f"""[Unit] Description = Routing to locally connected hardware After = network.target [Service] ExecStart = /usr/bin/python3 {TOOLS}/router.py [Install] WantedBy = multi-user.target """ class Frappy(Service): def __init__(self, cfg=None, port=None, requirements='', **kwds): if not cfg: self.servicecfg = None return main_ports.add(int(port or 10767)) if requirements: pip_requirements[cfg] = '\n'.join(requirements.split(',')) with open('/home/l_samenv/frappy/requirements.txt') as f: pip_requirements['frappy main'] = f.read() servicecfg = """[Unit] Description = Running frappy server # this waits also for network_online: After=repo-restore.service [Service] Type=simple Environment=PYTHONPATH=/home/l_samenv/.local/lib/python3.11/site-packages/ ExecStart = /home/l_samenv/boxtools/start_frappy [Install] WantedBy = default.target """ class Display(Service): as_root = True def __init__(self, **opts): if not opts: self.servicecfg = None servicecfg = f"""[Unit] Description = status display After = network.target [Service] User = root ExecStart = /usr/bin/python3 {TOOLS}/display.py -d [Install] WantedBy = multi-user.target """ class Boxweb(Service): def __init__(self, port=8080, wsport=8051, page=None): if page is None: self.servicecfg = None else: port = int(port) main_ports.add(port) wsport = int(wsport) main_ports.add(wsport) if port == 8080: main_ports.add(80) self.servicecfg = self.template.format(port=port, wsport=wsport, page=page) fixup = (TOOLS / 'to_config_desktop', Path('~/.config/systemd/user').expanduser(), False) if fixup not in box.fixup_dirs: # __init__ is called on every pass box.fixup_dirs.append(fixup) self.as_root = port <= 1024 template = """[Unit] Description = Web service for local GUI on a box After=repo-restore.service [Service] WorkingDirectory = /home/l_samenv/boxweb TimeoutStartSec = 60 # wait for frappy server serving websocket {wsport} is running ExecStartPre = /bin/sh -c 'until ss -H -ltn "sport = :{wsport}" | grep -q .; do sleep 1; done' ExecStart = /usr/bin/python3 flaskserver.py {page}.{port}.{wsport} [Install] WantedBy = default.target """ class RepoSync(Service): name = 'repo-sync' # the other units refer to this name def __init__(self, **opts): # no cfg section expected, but be tolerant for future extensions if not Path('/usr/bin/inotifywait').is_file(): installs.add('inotify-tools') servicecfg = """[Unit] Description=sync repos to central computer After=repo-restore.service [Service] ExecStart=/usr/bin/python3 %h/boxtools/repo_sync.py backup [Install] WantedBy=default.target """ class RepoRestore(RepoSync): name = 'repo-restore' # the other units refer to this name activate = False # only enabled: runs after boot (or when started by a unit depending on it) servicecfg = """[Unit] Description=Restore repos from central computer if requested Wants=network_online.service After=network_online.service [Service] Type=oneshot RemainAfterExit=yes ExecStart=/usr/bin/python3 %h/boxtools/repo_sync.py restore [Install] WantedBy=default.target """ pip_requirements = {} installs = set() box = BoxInfo() box.fixup_dirs = [ # (src, dst, as_root) (TOOLS / 'to_system', Path('/'), True), (TOOLS / f'to_system_{box.typ}', Path('/'), True), (TOOLS / 'to_home', Path.home(), False), ] box.hostname_changed = False dhcp_server_cfg = [] def do_cmd(*command, sudo=True): unix_cmd(*command, execute=doit, sudo=sudo) show.dirty = True def do_copy(src, dst): """copy or delete (src=None)""" as_root = not dst.is_relative_to('/home/l_samenv') if src is None: if dst.exists(): do_cmd('rm', dst, sudo=as_root) return if dst.exists() and getmtime(src) <= getmtime(dst): return do_cmd('cp', src, dst, sudo=as_root) def display_update(cfg): text = '\n'.join(cfg.get('startup_text', '').split('|')[:3]) text = text.replace('HOST', box.hostname) \ .replace('ADDR', box.macaddr) + '\n' if write_when_new(STARTUP_TEXT, text): print('change startup text') if doit: os.rename(STARTUP_TEXT, str(STARTUP_TEXT) + '.todo') return True return False def pip(): filename = TOOLS.parent / 'pip_requirements.txt' tmpname = TOOLS.parent / 'pip_requirements.tmp' pipcmd = f'pip3 install --user --break-system-packages -r {tmpname}' content = ''.join('# --- for %s ---\n%s\n' % kv for kv in pip_requirements.items()) if content: if write_when_new(filename, content, as_root=False): if doit: unix_cmd('mv', filename, tmpname, sudo=False) if os.system(pipcmd) == 0: unix_cmd('mv', tmpname, filename, sudo=False) else: unix_cmd('rm', tmpname, sudo=False) else: print(pipcmd) show.dirty = True SERVICES = [Router, Display, Frappy, Boxweb, RepoSync, RepoRestore] def update_host_key(hostname): oldkeysdir = None keys_changed = False pubpat = re.compile(r'(.*)@([^@]*)\n') try: found = False for pubfile in Path('/etc/ssh').glob('ssh_host*key.pub'): found = True pubhost = pubpat.match(pubfile.read_text()).group(2) if pubhost == hostname: continue # no need to change keys else: keys_changed = True prvfile = str(pubfile)[:-4] # strip '.pub' keysdir = f'/etc/ssh/old_keys_{pubhost}/' if keysdir != oldkeysdir: oldkeysdir = keysdir do_cmd('mkdir', '-p', keysdir) do_cmd('mv', pubfile, keysdir) do_cmd('mv', prvfile, keysdir) if keys_changed or not found: create_new = True for file in Path(f'/etc/ssh/old_keys_{hostname}/').glob('*'): do_cmd('mv', file, '/etc/ssh/') create_new = False if create_new: print('creating new ssh keys - this takes quite some time ...') do_cmd(f'ssh-keygen -A') # create missing keys for pubfile in Path('/etc/ssh').glob('ssh_host*key.pub'): head, name = pubpat.match(pubfile.read_text()).groups() if name != hostname and doit: do_cmd(f'chmod o+w {pubfile}') pubfile.write_text(f'{head}@{hostname}\n') do_cmd(f'chmod o-w {pubfile}') if keys_changed and doit: print('--- reboot needed ---') except BaseException as e: print(e) if oldkeysdir: for file in Path(oldkeysdir).glob('*'): do_cmd('mv', file, '/etc/ssh/') def write_when_new(filename, content, as_root=False, ignore_reduction=False): if content is None: lines = [] else: if not content.endswith('\n'): content += '\n' lines = content.split('\n') try: filepath = Path(filename) try: filepath.parent.mkdir(parents=True) # os.makedirs(Path(filename).parent, exist_ok=True) for parent in reversed(Path(filename).absolute().parents): unix_cmd('chmod', 'o+rx', parent, sudo=True) except FileExistsError: pass # do not complain when dir exists, and do not set mode old = filepath.read_text() except FileNotFoundError: old = None #except Exception as e: # print('can not read', filename, '-> do not check') # return False if old == content: return False if ignore_reduction: content_set = set(v for v in lines if not v.startswith('#')) old_set = set(v for v in (old or '').split('\n') if not v.startswith('#')) content_set -= old_set if content_set: print(f"missing in {filename}: {(', '.join(content_set - old_set))}") else: return False if doit: if lines: unix_cmd(f'mkdir -p {Path(filename).parent}', sudo=as_root) with tempfile.NamedTemporaryFile('w') as fil: fil.write(content) fil.flush() unix_cmd('cp', fil.name, filename, sudo=as_root) unix_cmd('chmod 0644', filename, sudo=as_root) else: unix_cmd('rm', '-f', filename, sudo=as_root) elif more_info: print('.' * 80) print('changes in', filename) old = [] if old is None else old.split('\n') top_lines = 0 # in case of empty loop for top_lines, (ol, ll) in enumerate(zip(old, lines)): if ol != ll: break bottom_lines = 0 for bottom_lines, (ol, ll) in enumerate(zip(reversed(old[top_lines:]), reversed(lines[top_lines:]))): if ol != ll: break if bottom_lines == 0: old.append('') lines.append('') bottom_lines += 1 print("===") print('\n'.join(old[:top_lines])) print('<<<') print('\n'.join(old[top_lines:-bottom_lines])) print('---') print('\n'.join(lines[top_lines:-bottom_lines])) print('>>>') print('\n'.join(old[-bottom_lines:-1])) print("===") print('.' * 80) else: print('write', filename) return content def replace_in_file(filename, pat, repl): """find pattern in content of filename find regexp pattern in content of file - if it does not match exactly once, an error is raised - exchange group 1 by repl - return True when result is changed, or False when not doit determines, whether the file is really changed """ with open(filename) as f: content = f.read() success = [] def fun(match, count=[]): if success: raise ValueError(f'{pat} matches multiple times in {filename}') part = match.group(0) match = match.re.match(part) fr, to = match.span(1) result = part[:fr] + repl + part[to:] if result != part: print(f'in {filename} replace {part} -> {result}') success.append(result != part) return result newcontent = re.sub(pat, fun, content) if not success: raise ValueError(f'{pat} not in {filename}') write_when_new(filename, newcontent, as_root=True) return success[0] def create_if(name, cfg): if cfg == 'off': result = None elif cfg.startswith('wan') or cfg == 'dhcp': if box.main_if != name: print(box.main_if, name) raise ValueError('can not have more than one WAN/DHCP port') box.main_if = name # default: all <= 192.0.0.0 # others have to be added explicitly dhcp_server_cfg.append(('0.0.0.0/128.0.0.0', [])) dhcp_server_cfg.append(('128.0.0.0/192.0.0.0', [])) for nw in cfg.split(',')[1:]: nw = IPv4Interface(nw).network dhcp_server_cfg.append((nw.with_netmask, [])) result = f"allow-hotplug {name}\niface {name} inet dhcp" else: cfgip = IPv4Interface(cfg) network = cfgip.network if network.prefixlen == 32: # or no prefix specified otherip = IPv4Interface('%s/24' % cfgip.ip) network = otherip.network if str(cfgip.ip).endswith('.1'): cfgip = network.network_address + 2 else: cfgip = network.network_address + 1 dhcp_server_cfg.append((network.with_netmask, [(str(otherip.ip), str(otherip.ip))])) addr = str(cfgip) else: # subnet with multiple adresses -> static adresses only. dhcp range not yet implemented addr = str(cfgip.ip) # result['NETMASK'] = network.netmask result = f"allow-hotplug {name}\niface {name} inet static\n address {addr}/{network.prefixlen}" if result: return result + '\n' def walk_dir(action, srcpath, dstroot, files): if not files: return action.srcpath = srcpath = Path(srcpath) action.dstpath = dstpath = dstroot / srcpath match, mismatch, missing = filecmp.cmpfiles(srcpath, dstpath, files) if mismatch: newer = [f for f in mismatch if getmtime(dstpath / f) > getmtime(srcpath / f)] if newer: action.newer(newer) if len(newer) < len(mismatch): newer = set(newer) action.older([f for f in mismatch if f not in newer]) if missing: if DEL in missing: missing.remove(DEL) with open(srcpath / DEL) as fil: to_delete = [] for fname in fil: fname = fname.strip() if fname and exists(dstpath / fname): if (srcpath / fname).exists(): print('ERROR: %s in %s, but also in repo -> ignored' % (fname, DEL)) else: to_delete.append(fname) action.delete(to_delete) if missing: action.missing(missing) def walk(action): for src, dst, as_root in box.fixup_dirs: if src.exists(): action.as_root = as_root os.chdir(src) for srcpath, _, files in os.walk('.'): walk_dir(action, srcpath, dst, files) class Walker: srcpath = None dstpath = None as_root = None class Show(Walker): dirty = False def diff(self, title, files): self.dirty = True if more_info: for f in files: if f.endswith(more_info): print('diff %s %s' % (self.srcpath / f, self.dstpath / f)) print('.' * 80) else: print('%s %s:\n %s' % (title, self.dstpath, ' '.join(files))) def show(self, title, srcpath, files): self.dirty = True if more_info: for f in files: print('cat %s' % (srcpath / f)) print('.' * 80) else: print('%s %s:\n %s' % (title, self.dstpath, ' '.join(files))) def newer(self, files): self.show('get from', self.srcpath, files) def older(self, files): self.show('replace in', self.srcpath, files) def missing(self, files): self.show('install in', self.srcpath, files) def delete(self, files): self.show('remove from', self.dstpath, files) class Do(Walker): def newer(self, files): unix_cmd('mkdir', '-p', str(self.dstpath), sudo=self.as_root) unix_cmd('cp', *(str(self.srcpath / f) for f in files), str(self.dstpath), sudo=self.as_root) older = newer missing = newer def delete(self, files): unix_cmd('rm', '-f', *(str(self.dstpath / f) for f in files)) def handle_config(): dhcp_server_cfg.clear() try: config = box.read_config() print(f'{box.cfgfile} read') except UndefinedConfigFile as e: print(f'{box.cfgfile} not found', e) if box.oldcfg: converted = convert_cfg(box.cfgfile, box.macaddr) print(f'convert {box.cfgfile} to {converted}') box.cfgfile = converted box.oldcfg = False # else the conversion would be repeated in the next call config = box.read_config() cfgfile = box.cfgfile if cfgfile: unix_cmd('scp', str(cfgfile), f'{CENTRAL_BOXTOOLS_REPO}/cfg/', sudo=False) newhostname = cfgfile.stem if cfgfile else None if cfgfile: if box.hwtype != 'apu': typ = config.get('BOX', {}).get('type') if typ: if typ not in BOX_TYPES: raise ValueError(f'unknown type={typ} defined in BOX section') box.typ = typ elif box.typ == 'cm4': print('WARNING: missing type in BOX section') print('This is a cm4 so guess its a dual-eth-rpi') box.typ = 'dual-eth-rpi' elif box.typ == 'cm3': print('WARNING: missing type in BOX section') print('This is a cm3 so guess its a ionopimax') box.typ = 'ionopimax' print('box type:', box.typ, ' mac addr:', box.macaddr) print('---') else: template_args = {'mac': box.macaddr} if box.typ == 'apu': # determine if display is present disp = serial.Serial('/dev/ttyS1', baudrate=115200, timeout=1) disp.write(b'\x1b\x1b\x01\xf3') display_available = disp.read(8)[0:4] == b'\x1b\x1b\x05\xf3' if display_available: box.typ = 'control-box' else: box.typ = 'bare-apu' elif box.typ == 'cm4': box.typ = 'dual-eth-rpi' print('This is a cm4, so guess its a dual-eth-rpi - please check type in created cfg file') template_args['mac2'] = 'enx?' for addr in box.network_interfaces: if addr.startswith('enx'): template_args['mac2'] = addr print(template_args) elif box.typ == 'cm3': print('This is a cm3 so guess its a ionopimax - please check type in created cfg file') box.typ = 'ionopimax' elif box.typ == 'rpi': print('This is a simple rpi, so guess its a ionopi - please check type in created cfg file') box.typ = 'ionopi' template = TEMPLATES.get(box.typ) if template is None: box.typ = box.typ or 'unknown-box' template = GENERIC_TEMPLATE template_args['typ'] = box.typ template = template % template_args print('no cfg file found for this', box.typ, f'with mac address {box.macaddr} (hostname={box.hostname})') newhostname = input('enter host name: ') if not newhostname: print('no hostname given') return False cfgfile = BoxInfo.CFGDIR / f'{newhostname}.cfg' with open(cfgfile, 'w') as f: f.write(template) config = box.read_config() cfgfile = box.cfgfile if box.hostname_changed or box.hostname != newhostname: box.hostname_changed = True if cfgfile: newhostname = cfgfile.stem.split('_')[0] if doit: print('bash sethostname.sh') unix_cmd('bash', f'{TOOLS}/sethostname.sh') else: if cfgfile: print('replace host name %r by %r' % (box.hostname, newhostname)) show.dirty = True config = box.read_config() cfgpath = Path(box.cfgfile) cfglink = Path.home() / 'box.cfg' if not cfglink.exists() or cfglink.resolve() != cfgpath: print(f'ln -s {cfgpath} {cfglink}') if doit: cfglink.symlink_to(cfgpath) else: print(f'ln -s {cfgpath} {cfglink}') box.hostname = newhostname update_host_key(newhostname) if cfgfile is None: return False to_start = {} # dict of , ifname = '' try: netcfg = config.get('NETWORK', {}) if netcfg: # when no network is specified, do not handle network at all for name in netcfg: if name not in box.network_interfaces: print(f'{name} is currently not a valid network interface - skip') continue for ifname in box.network_interfaces: content = create_if(ifname, netcfg.get(ifname, 'off')) # content = '\n'.join('%s=%s' % kv for kv in content.items()) todo = write_when_new(f'/etc/network/interfaces.d/{ifname}', content, as_root=True) if todo and not more_info: print('change', ifname) show.dirty = True to_start[ifname] = 'if_restart', True if dhcp_server_cfg: content = [DHCP_HEADER] for subnet, rangelist in dhcp_server_cfg: try: adr, mask = subnet.split('/') except Exception as e: print(subnet, repr(e)) continue content.append('subnet %s netmask %s {\n' % (adr, mask)) #content.append(' option netmask %s;\n' % mask) for rng in rangelist: content.append(' range %s %s;\n' % rng) content.append('}\n') content = ''.join(content) todo = write_when_new('/etc/dhcp/dhcpd.conf', content, as_root=True) if todo: print('change dhcpd.conf') to_start['isc-dhcp-server'] = 'restart', True show.dirty = True try: if replace_in_file( '/etc/default/isc-dhcp-server', r'INTERFACESv4="(.*)"', ' '.join( [n for n in box.network_interfaces if n != box.main_if])): show.dirty = True except FileNotFoundError: if 'yes'.startswith(input('install isc-dhcp-server? [y]')): unix_cmd('apt-get install isc-dhcp-server') exit() elif doit: unix_cmd('systemctl stop isc-dhcp-server') unix_cmd('systemctl disable isc-dhcp-server') displaycfg = config.get('DISPLAY') if displaycfg and display_update(displaycfg): to_start['display'] = 'restart', True if change_to_gitea(doit): show.dirty = True except Exception as e: print('ERROR: can not handle %s %s: %r' % (box.hostname, ifname, e)) raise # return False reload_systemd = set() # (re)build the services here, as the config might have changed above # (new cfg file, host name change), and constructing them registers ports and packages for servicecls in SERVICES: section_dict = config.get(servicecls.__name__.upper(), {}) servicecls(**section_dict).handle(to_start, reload_systemd) if change_firewall(router_ports, main_ports, doit): show.dirty = True if 'dialout' not in unix_cmd('id l_samenv'): do_cmd('usermod -a -G dialout l_samenv') if installs: do_cmd('apt-get -y update') for pkg in installs: do_cmd(f'apt-get -y install {pkg}') if pip_requirements: pip() for as_root in reload_systemd: do_cmd('systemctl daemon-reload', sudo=as_root) # '--user' will be inserted inside for service, (action, _) in to_start.items(): show.dirty = True if action == 'if_restart': do_cmd(f'ifconfig {service} down') for service, (action, as_root) in to_start.items(): if action == 'if_restart': do_cmd(f'ifconfig {service} up') else: if action == 'restart': # else 'enable' do_cmd('systemctl', 'restart', service, sudo=as_root) if not as_root: do_cmd('loginctl enable-linger', sudo=True) do_cmd('systemctl', 'enable', service, sudo=as_root) if box.change_if_names: print('interface name system has to be changed from enp*s0 to eth*') if replace_in_file('/etc/default/grub', r'GRUB_CMDLINE_LINUX_DEFAULT="quiet(.*)"', ' net.ifnames=0'): do_cmd('update-grub') result = [f'config file:\n {cfgfile}'] for section, section_dict in config.items(): result.append(section) for key, value in section_dict.items(): result.append(f' {key} = {value}') result.append('') return '\n'.join(result) doit = False print(' ') show = Show() result = handle_config() walk(show) if not result: print('fix first above errors') else: if show.dirty: print('---') answer = input('enter "y(es)" to do above or "m(ore)" for more info: ') if not answer: print('cancel') elif 'yes'.startswith(answer.lower()): doit = True handle_config() walk(Do()) unix_cmd(f'rm -f {TOOLS / "current"}') with open(TOOLS / 'current', 'w') as f: f.write(result) elif 'more'.startswith(answer.lower()): more_info = True walk(show) result = handle_config() else: print('nothing to do')