diff --git a/.github/workflows/genesis_openembedded.yml b/.github/workflows/genesis_openembedded.yml index 54b71d9d1..e025a4d76 100644 --- a/.github/workflows/genesis_openembedded.yml +++ b/.github/workflows/genesis_openembedded.yml @@ -22,7 +22,7 @@ jobs: sudo env DEBIAN_FRONTEND=noninteractive apt-get install -y \ build-essential meson ninja-build pkg-config libnewt-dev \ libsystemd-dev python3-pyte bubblewrap systemd-journal-remote - - name: Enable isolated journal test namespaces + - name: Enable isolated console test namespaces run: sudo sysctl -w kernel.apparmor_restrict_unprivileged_userns=0 - name: Exercise the newt dialogs run: /usr/bin/python3 xCAT-test/native/genesis_console_dialogs.py diff --git a/xCAT-test/native/genesis_console_dialogs.py b/xCAT-test/native/genesis_console_dialogs.py index 43f78cf1d..16e65e693 100644 --- a/xCAT-test/native/genesis_console_dialogs.py +++ b/xCAT-test/native/genesis_console_dialogs.py @@ -7,6 +7,7 @@ import os from pathlib import Path import pty import select +import shlex import signal import struct import subprocess @@ -61,6 +62,16 @@ class ConsoleTerminal(unittest.TestCase): def console_command(self): return [str(self.binary)] + def isolated_prefix(self, mounts): + command = ["bwrap", "--unshare-all", "--die-with-parent", "--ro-bind", "/", "/", + "--dev", "/dev", "--proc", "/proc", "--tmpfs", "/run", "--tmpfs", "/var/log", + *mounts] + result = subprocess.run(command + ["/bin/true"], text=True, + capture_output=True, timeout=10) + self.assertEqual(result.returncode, 0, + "Console tests require working Bubblewrap namespaces: " + result.stderr) + return command + def stop(self): if self.process.poll() is None: self.process.terminate() @@ -196,19 +207,12 @@ class ConsoleJournal(ConsoleTerminal): self.entries = [] machine_id_file = self.root / "machine-id" machine_id_file.write_text(self.machine_id + "\n") - sandbox = [ - "bwrap", "--unshare-all", "--die-with-parent", "--ro-bind", "/", "/", - "--dev", "/dev", "--proc", "/proc", "--tmpfs", "/run", "--tmpfs", "/var/log", + self.sandbox = self.isolated_prefix([ "--bind", str(self.root), str(self.root), "--ro-bind", str(machine_id_file), "/etc/machine-id", "--ro-bind", str(journal_root), "/var/log/journal", - ] - result = subprocess.run(sandbox + ["/bin/true"], text=True, - capture_output=True, timeout=10) - self.assertEqual(result.returncode, 0, - "Journal tests require working Bubblewrap namespaces: " + result.stderr) - self.sandbox = sandbox - return sandbox + [str(self.binary)] + ]) + return self.sandbox + [str(self.binary)] def write_journal(self, entries, interval=1000000): records = [] @@ -474,5 +478,113 @@ class ConsoleJournal(ConsoleTerminal): self.wait_text("RECORD-0129") +class ConsoleShell(ConsoleTerminal): + """Mount a fixture libexec directory containing the external shell launcher. + + The linked console still owns confirmation, fork/exec, waiting and redraw. + The per-test temporary root is the only writable host-backed directory. + The libexec mount is read-only. + /run and /var/log are private tmpfs mounts. + """ + + def console_command(self): + directory = self.root / "libexec" + (directory / "xcat").mkdir(parents=True) + self.launcher = directory / "xcat/genesis-maintenance-shell" + self.calls = self.root / "shell-calls" + self.input = self.root / "shell-input" + self.write_launcher(0) + return self.isolated_prefix([ + "--bind", str(self.root), str(self.root), + "--ro-bind", str(directory), "/usr/libexec", + ]) + [str(self.binary)] + + def write_launcher(self, exit_status): + self.launcher.write_text( + "#!/bin/sh\n" + f"printf '%s\\n' started >> {shlex.quote(str(self.calls))}\n" + "printf 'SHELL-RUNNING\\n'\n" + "IFS= read -r answer\n" + f"printf '%s\\n' \"$answer\" > {shlex.quote(str(self.input))}\n" + f"exit {exit_status}\n") + self.launcher.chmod(0o755) + + def shell_dialog_text(self, display): + lines = display.splitlines() + for row, line in enumerate(lines): + if "Maintenance shell" not in line: + continue + left, right = line.rfind("┌"), line.find("┐") + if left < 0 or right <= left: + continue + content = [] + for line in lines[row + 1:]: + if line[left:right + 1].startswith("└") and line[right] == "┘": + return self.visible_text("\n".join(content)) + content.append(line[left + 1:right]) + return "" + + def confirm_shell(self): + self.key(b"\x1b[24~") + display = self.wait_screen( + lambda text: "Open a root maintenance shell?" in self.shell_dialog_text(text), + "maintenance shell confirmation") + self.assertIn("Provisioning services continue in the background.", + self.shell_dialog_text(display)) + + def wait_main(self): + def is_main(display): + return ("Maintenance shell" not in display + and "SHELL-RUNNING" not in display + and "Genesis diagnostics" not in display + and "Genesis status" in display) + + self.wait_screen(is_main, "the restored status console") + self.assert_screen_stays(is_main, "The restored status console changed", 1) + + def test_cancel_does_not_launch_shell(self): + self.confirm_shell() + self.assertFalse(self.calls.exists()) + self.key(b"\t\r") + self.wait_main() + self.assertFalse(self.calls.exists()) + self.key(b"\x1bOP") + self.wait_text("Genesis help") + + def test_confirmed_shell_accepts_input_and_returns_to_console(self): + for count, exit_status in enumerate((0, 7), 1): + with self.subTest(launch=count, exit_status=exit_status): + self.write_launcher(exit_status) + self.confirm_shell() + self.assertEqual(self.calls.read_text() if self.calls.exists() else "", + "started\n" * (count - 1)) + self.key(b"\r") + self.wait_text("SHELL-RUNNING") + self.assertEqual(self.calls.read_text(), "started\n" * count) + self.key(f"operator-input-{count}\r".encode()) + self.wait_main() + self.assertEqual(self.input.read_text(), f"operator-input-{count}\n") + self.assertIsNone(self.process.poll()) + self.key(b"\x1bOQ") + self.wait_text("Genesis diagnostics") + self.key(b"\x1bOQ") + self.wait_main() + + def test_shell_exec_error_is_reported_and_console_recovers(self): + self.launcher.unlink() + self.confirm_shell() + self.key(b"\r") + display = self.wait_screen( + lambda text: "No such file or directory" in self.shell_dialog_text(text), + "the maintenance shell executable error") + self.assertIn("The maintenance shell could not be opened: No such file or directory", + self.shell_dialog_text(display)) + self.assertFalse(self.calls.exists()) + self.key(b"\r") + self.wait_main() + self.key(b"\x1bOP") + self.wait_text("Genesis help") + + if __name__ == "__main__": unittest.main(verbosity=2)