rework display.py: need to be tested!

.gitignore

@@ -1,3 +1,5 @@
 current
 startup_display.txt
 __pycache__
+boot_tiny
+display_firmware

display.py

@@ -13,6 +13,8 @@ from configparser import ConfigParser
 tty = '/dev/ttyS1'
 TOOLS = '/home/l_samenv/boxtools'
 STARTUP_TEXT = f'{TOOLS}/startup_display.txt'
+FW_SRC = 'l_samenv@linse-c:boxes/display_firmware'
+FW_DIR = 'display_fw'
 
 ESC = 0x1b
 ESCESC = bytes((ESC, ESC))
@@ -253,14 +255,14 @@ def pretty_version(v):
 
 
 daemon = False
-firmware = None
+firmware = False
 
 for arg in sys.argv[1:]:
     if arg == '-d':
         daemon = True
-    elif arg.endswith('.bin'):
-        firmware = arg
-    elif tty.startswith('/dev/'):
+    elif arg == '-f':
+        firmware = True
+    elif arg.startswith('/dev/'):
         tty = arg
     else:
         raise ValueError(f'do not know how to treat argument: {arg}')
@@ -271,21 +273,32 @@ if not firmware:
 
 d = Display(tty, daemon=daemon and not firmware)
 if firmware:
-    with open(firmware, 'rb') as f:
-        tag = f.read()[-8:]
-    if tag[:4] != MAGIC:
-        raise ValueError(f'{firmware} is not a valid firmware file')
-    hwversion = d.version()
-    if tag[4:] != hwversion:
-        print('display version:', pretty_version(hwversion))
-        print('binfile version:', pretty_version(tag[4:]))
-        result = input('flash this (takes 1 min)? ').lower()
-        if result in ('y', 'yes'):
-            print('\ndo NOT interrupt')
-            d.bootmode()
-            d.term.close()
-            time.sleep(1.)
-            os.system(f'../stm32flash-0.7/stm32flash -R -v -b 115200 -w {firmware} {tty}')
+    os.system(f'mkdir -p {FW_DIR}')
+    os.system(f'rm -rf {FW_DIR}/*')
+    try:
+        os.system(f'scp {FW_SRC}/fw_display.bin {FW_DIR}/fw_display.bin')
+        fwbin = f'{FW_DIR}/fw_display.bin'
+        print('FWBIN', fwbin)
+        with open(fwbin, 'rb') as f:
+            tag = f.read()[-8:]
+        if tag[:4] != MAGIC:
+            raise ValueError(f'{fwbin} is not a valid firmware file')
+        hwversion = d.version()
+        if tag[4:] != hwversion:
+            print('display version:', pretty_version(hwversion))
+            print('binfile version:', pretty_version(tag[4:]))
+            result = input('flash this (takes 1 min)? ').lower()
+            if result in ('y', 'yes'):
+                print('\ndo NOT interrupt')
+                d.bootmode()
+                d.term.close()
+                time.sleep(1.)
+                os.system(f'scp -r {FIRMWARE_SRC}/stm32flash-0.7 {FW_DIR}/')
+                os.system(f'{FW_DIR}/stm32flash-0.7/stm32flash -R -v -b 115200 -w {fwbin} {tty}')
+    finally:
+        print('cleanup')
+        os.system(f'rm -rf {FW_DIR}')
+        print('cleanup')
 if daemon:
     while True:
         d.refresh()

ssh_cert_login.py

@@ -0,0 +1,233 @@
+"""Make this box accept SSH certificate logins (signed by our SSH CA).
+
+Use from your main script:
+
+    from provision_ssh_ca import install_ssh_cert_login
+
+    install_ssh_cert_login(doit=False)   # only report what would be done
+    install_ssh_cert_login(doit=True)    # do it
+
+Run as a normal user: system commands go through unix_cmd (with sudo), the CA
+public keys are fetched with scp as the calling user (own ssh key/agent).
+A command failed if unix_cmd returns a Failure (non-zero return code, or the
+command could not be started). With doit=False unix_cmd only prints and
+returns an empty string.
+
+What it installs (idempotent, only differences are touched):
+
+  /etc/ssh/ca.pub                              all CA public keys
+  /etc/ssh/principals/<account>                principals allowed for <account>
+  /etc/ssh/sshd_config.d/10-ca.conf            TrustedUserCAKeys + AuthorizedPrincipalsFile
+  /etc/ssh/sshd_config.d/20-no-password.conf   only if DISABLE_PASSWORD = True
+
+Before sshd is reloaded the configuration is checked with `sshd -t`; if the
+check fails, all changes are rolled back. Reloading does not close existing
+sessions. Existing authorized_keys and password logins keep working unless
+DISABLE_PASSWORD is set.
+"""
+import os
+import re
+import tempfile
+from pathlib import Path
+
+from utils import unix_cmd, Failure
+
+# ---- configuration ---------------------------------------------------------
+CENTRAL_HOST = "linse-c"
+CENTRAL_USER = "l_samenv"       # None: your usual ssh login name / ssh config
+CENTRAL_CA_DIR = "boxes/ssh_cert_login"    # must be readable by normal users
+CA_FILES = ["ssh_ca.pub", "ssh_ca_spare.pub"]   # public keys only!
+
+ACCOUNTS = ["l_samenv"]         # local login accounts people use on the box
+PRINCIPALS = ["l_samenv-admin"] # principals allowed to log in to those accounts
+DISABLE_PASSWORD = False        # set True only after cert login is tested
+# ----------------------------------------------------------------------------
+
+SSH_DIR = Path("/etc/ssh")
+MAIN_CONF = SSH_DIR / "sshd_config"
+DROPIN_DIR = SSH_DIR / "sshd_config.d"
+PRINCIPALS_DIR = SSH_DIR / "principals"
+CA_FILE = SSH_DIR / "ca.pub"
+KEY_PREFIXES = ("ssh-", "ecdsa-", "sk-")
+SYSTEMD_DIRS = ("/etc/systemd/system", "/lib/systemd/system",
+                "/usr/lib/systemd/system")
+
+
+class _Problem(Exception):
+    pass
+
+
+def _say(msg):
+    print(f"ssh cert login: {msg}")
+
+
+def _sshd():
+    for p in ("/usr/sbin/sshd", "/usr/local/sbin/sshd"):
+        if os.path.exists(p):
+            return p
+    return "sshd"
+
+
+def _ssh_unit():
+    """Name of the systemd unit of the ssh server: ssh (Debian) or sshd."""
+    for unit in ("ssh", "sshd"):
+        if any(os.path.exists(f"{d}/{unit}.service") for d in SYSTEMD_DIRS):
+            return unit
+    return "ssh"
+
+
+def _check_include():
+    try:
+        text = MAIN_CONF.read_text()
+    except OSError as e:
+        raise _Problem(f"cannot read {MAIN_CONF}: {e}")
+    for ln in text.splitlines():
+        s = ln.strip()
+        if s.startswith("Include") and "sshd_config.d" in s:
+            return
+    raise _Problem(
+        f"{MAIN_CONF} has no 'Include /etc/ssh/sshd_config.d/*.conf' line. "
+        "Add it at the top of the file (sshd uses the first value it sees "
+        "for each option), then try again.")
+
+
+def _fetch_ca_keys(dest):
+    """scp the CA public keys from the central machine into dest (silent)."""
+    target = f"{CENTRAL_USER}@{CENTRAL_HOST}" if CENTRAL_USER else CENTRAL_HOST
+    sources = [f"{target}:{CENTRAL_CA_DIR}/{name}" for name in CA_FILES]
+    res = unix_cmd("scp", "-q", *sources, str(dest), sudo=False, execute=None)
+    missing = [n for n in CA_FILES if not (Path(dest) / n).is_file()]
+    if isinstance(res, Failure) or missing:
+        raise _Problem(f"cannot fetch {', '.join(missing) or 'CA keys'} from "
+                       f"{target}:{CENTRAL_CA_DIR} (is the central machine reachable?)")
+
+
+def _read_ca_keys(path):
+    """Return the public key lines of a CA file, refusing anything suspicious."""
+    text = path.read_text()
+    if "PRIVATE KEY" in text:
+        raise _Problem(f"{path.name} contains a PRIVATE key; refusing to use it")
+    lines = [ln.strip() for ln in text.splitlines()
+             if ln.strip() and not ln.lstrip().startswith("#")]
+    if not lines:
+        raise _Problem(f"{path.name}: no keys found")
+    for ln in lines:
+        if not ln.startswith(KEY_PREFIXES) or "-cert-v01" in ln.split()[0]:
+            raise _Problem(f"{path.name}: not a plain public key line: {ln[:40]}...")
+    out = unix_cmd("ssh-keygen", "-l", "-f", str(path), sudo=False, execute=None)
+    if isinstance(out, Failure) or len(out.strip().splitlines()) != len(lines):
+        raise _Problem(f"{path.name}: ssh-keygen does not accept this file")
+    return lines
+
+
+def _wanted(keys):
+    """Files to have on the box: {path: content}; the drop-in comes last."""
+    wanted = {CA_FILE: "\n".join(keys) + "\n"}
+    for acct in ACCOUNTS:
+        wanted[PRINCIPALS_DIR / acct] = "\n".join(PRINCIPALS) + "\n"
+    if DISABLE_PASSWORD:
+        wanted[DROPIN_DIR / "20-no-password.conf"] = (
+            "# managed by provision_ssh_ca\n"
+            "PasswordAuthentication no\n"
+            "KbdInteractiveAuthentication no\n")
+    wanted[DROPIN_DIR / "10-ca.conf"] = (
+        "# managed by provision_ssh_ca\n"
+        f"TrustedUserCAKeys {CA_FILE}\n"
+        f"AuthorizedPrincipalsFile {PRINCIPALS_DIR}/%u\n")
+    return wanted
+
+
+def _put(src, dest, doit):
+    return unix_cmd("install", "-o", "root", "-g", "root", "-m", "644",
+                    str(src), str(dest), execute=doit)
+
+
+def _rollback(done, tmp):
+    _say("rolling back")
+    for i, path, old in reversed(done):
+        if old is None:
+            unix_cmd("rm", "-f", str(path), execute=True)
+        else:
+            src = tmp / f"old{i}"
+            src.write_text(old)
+            _put(src, path, True)
+
+
+def _install(doit):
+    for name in ACCOUNTS + PRINCIPALS:
+        if not re.fullmatch(r"[A-Za-z0-9_.-]+", name):
+            raise _Problem(f"unsafe name in configuration: {name!r}")
+    _check_include()
+
+    with tempfile.TemporaryDirectory() as tmpname:
+        tmp = Path(tmpname)
+        _fetch_ca_keys(tmp)
+        keys = []
+        for name in CA_FILES:
+            for k in _read_ca_keys(tmp / name):
+                if k not in keys:
+                    keys.append(k)
+
+        todo = []                       # (path, new content, old content or None)
+        for path, content in _wanted(keys).items():
+            try:
+                old = path.read_text() if path.exists() else None
+            except OSError as e:
+                raise _Problem(f"cannot read {path}: {e}")
+            if old != content:
+                todo.append((path, content, old))
+        if not todo:
+            return True                 # up to date: stay quiet
+
+        _say("installing:" if doit else "to do:")
+        dirs = sorted({p.parent for p, _, _ in todo if not p.parent.is_dir()})
+        for d in dirs:
+            _say(f"create {d}")
+            res = unix_cmd("install", "-d", "-o", "root", "-g", "root",
+                           "-m", "755", str(d), execute=doit)
+            if isinstance(res, Failure):
+                raise _Problem(f"cannot create {d}")
+
+        done = []                       # (index, path, old content) for rollback
+        for i, (path, new, old) in enumerate(todo):
+            _say(f"{'create' if old is None else 'update'} {path}")
+            src = tmp / f"new{i}"
+            src.write_text(new)
+            res = _put(src, path, doit)
+            done.append((i, path, old))
+            if isinstance(res, Failure):
+                _rollback(done, tmp)
+                raise _Problem(f"cannot install {path}")
+
+        _say("check configuration")
+        res = unix_cmd(_sshd(), "-t", execute=doit)
+        if isinstance(res, Failure):
+            _rollback(done, tmp)
+            raise _Problem("sshd rejected the new configuration (see message above)")
+
+        unit = _ssh_unit()
+        _say("reload sshd")
+        res = unix_cmd("systemctl", "reload", unit, execute=doit)
+        if isinstance(res, Failure):
+            raise _Problem(f"configuration is valid, but 'systemctl reload {unit}' failed")
+        if doit:
+            _say("done; test certificate login from another terminal "
+                 "before closing this session")
+        return doit
+
+
+def install(doit):
+    """Set up SSH certificate login on this box.
+
+    doit=False: change nothing; stay quiet if all is in place, else print
+                what has to be done.
+    doit=True:  do what is needed and print the commands.
+
+    Returns True if the box is set up when the function returns, else False
+    (changes pending, or a problem, which is printed).
+    """
+    try:
+        return _install(doit)
+    except _Problem as e:
+        _say(f"ERROR: {e}")
+        return False