From 9f257a9e97d72d67f70f36b30a183d1f97f58b9a Mon Sep 17 00:00:00 2001 From: Markus Zolliker Date: Thu, 29 Oct 2026 15:43:29 +0100 Subject: [PATCH] rework display.py: need to be tested! --- .gitignore | 2 + display.py | 51 ++++++---- ssh_cert_login.py | 233 ++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 267 insertions(+), 19 deletions(-) create mode 100644 ssh_cert_login.py diff --git a/.gitignore b/.gitignore index 0d2e966..6f26885 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,5 @@ current startup_display.txt __pycache__ +boot_tiny +display_firmware diff --git a/display.py b/display.py index 544ca01..2e74919 100644 --- a/display.py +++ b/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() diff --git a/ssh_cert_login.py b/ssh_cert_login.py new file mode 100644 index 0000000..3b55a79 --- /dev/null +++ b/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/ principals allowed for + /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