From 7999fab25c5172207ee41e38732f4923565ff6b6 Mon Sep 17 00:00:00 2001 From: Andrew Guschin Date: Fri, 30 Oct 2026 04:43:59 +0400 Subject: Validate controller diagnostics, reboot protection, and terminal authentication --- tests/vm_controller_acceptance.py | 126 +++++++++++++++++++++++++++++++++++++- 1 file changed, 125 insertions(+), 1 deletion(-) (limited to 'tests/vm_controller_acceptance.py') diff --git a/tests/vm_controller_acceptance.py b/tests/vm_controller_acceptance.py index 1d3405a..a6b6bc4 100644 --- a/tests/vm_controller_acceptance.py +++ b/tests/vm_controller_acceptance.py @@ -13,6 +13,8 @@ import termios import fcntl import struct import pwd +import signal +import secrets import subprocess import threading import time @@ -38,6 +40,119 @@ def cli(*args, **kwargs): return command(BIN, *args, **kwargs) def status(): return json.loads(cli('status', '--json').stdout) +def cancel_terminal_authorization(): + before = status() + process, terminal = pty.fork() + if process == 0: + user = pwd.getpwnam('rustacean') + os.initgroups(user.pw_name, user.pw_gid) + os.setgid(user.pw_gid); os.setuid(user.pw_uid) + os.environ.update(LANG='C', LC_ALL='C', TMPDIR='/var/tmp') + os.execv(BIN, [BIN, 'configuration', 'unset']) + output = b'' + reaped = False + try: + deadline = time.monotonic() + 10 + while b'Password:' not in output and time.monotonic() < deadline: + if select.select([terminal], [], [], 0.2)[0]: + try: output += os.read(terminal, 65536) + except OSError: break + assert b'Password:' in output, f'Terminal authentication did not prompt: {output.decode(errors="replace")}' + os.write(terminal, b'\x03') + deadline = time.monotonic() + 5 + while time.monotonic() < deadline: + pid, result = os.waitpid(process, os.WNOHANG) + if pid: + reaped = True + assert os.waitstatus_to_exitcode(result) != 0 + break + time.sleep(0.05) + assert reaped, 'Cancelled terminal authentication did not exit' + after = status() + assert after['selected_configuration'] == before['selected_configuration'] + assert after['requested_enabled'] == before['requested_enabled'] + assert after['generation'] == before['generation'] + finally: + with contextlib.suppress(ProcessLookupError): os.killpg(process, signal.SIGTERM) + if not reaped: + with contextlib.suppress(ChildProcessError): os.waitpid(process, 0) + os.close(terminal) + + +def terminal_password_attempt(password, succeeds): + process, terminal = pty.fork() + if process == 0: + user = pwd.getpwnam('rustacean') + os.initgroups(user.pw_name, user.pw_gid) + os.setgid(user.pw_gid); os.setuid(user.pw_uid) + os.environ.update(LANG='C', LC_ALL='C', TMPDIR='/var/tmp') + os.execv(BIN, [BIN, 'configuration', 'unset']) + output = b'' + reaped = False + try: + deadline = time.monotonic() + 12 + while b'Password:' not in output and time.monotonic() < deadline: + if select.select([terminal], [], [], 0.2)[0]: + try: output += os.read(terminal, 65536) + except OSError: break + assert b'Password:' in output, 'The VM fixture admin did not receive a terminal password prompt' + time.sleep(0.1) + assert not termios.tcgetattr(terminal)[3] & termios.ECHO, 'Password input must not echo' + os.write(terminal, password.encode() + b'\n') + if not succeeds: + observed = b'' + while time.monotonic() < deadline: + if select.select([terminal], [], [], 0.2)[0]: + try: observed += os.read(terminal, 65536) + except OSError: break + if b'failed' in observed.lower() or b'failure' in observed.lower() or b'Password:' in observed: break + assert b'failed' in observed.lower() or b'failure' in observed.lower() or b'Password:' in observed, 'Wrong-password rejection was not observed' + os.write(terminal, b'\x03') + while time.monotonic() < deadline: + if select.select([terminal], [], [], 0.1)[0]: + with contextlib.suppress(OSError): os.read(terminal, 65536) + pid, result = os.waitpid(process, os.WNOHANG) + if pid: + reaped = True + assert (os.waitstatus_to_exitcode(result) == 0) == succeeds + return + raise AssertionError('Terminal authentication did not complete within the deadline') + finally: + with contextlib.suppress(ProcessLookupError): os.killpg(process, signal.SIGTERM) + if not reaped: + with contextlib.suppress(ChildProcessError): os.waitpid(process, 0) + os.close(terminal) + + +def password_authorization_fixture(): + name = 'roentgen-auth-fixture' + try: + pwd.getpwnam(name) + except KeyError: pass + else: raise AssertionError('Refusing to modify an existing VM fixture account') + policy = Path('/etc/polkit-1/rules.d/00-000-roentgen-auth-fixture.rules') + command('useradd', '--no-create-home', '--shell', '/bin/sh', name) + try: + # This temporary VM-only password is test data. It never enters argv, + # logs, repository files or the real user's authentication records. + password = secrets.token_urlsafe(24) + command('chpasswd', input=name + ':' + password + '\n') + policy.write_text('polkit.addAdminRule(function(action, subject) { if (action.id == "org.roentgen.control" && subject.user == "rustacean") return ["unix-user:roentgen-auth-fixture"]; });\n') + policy.chmod(0o644); time.sleep(1) + before = status() + terminal_password_attempt('deliberately-wrong-fixture-password', succeeds=False) + after = status() + assert after['generation'] == before['generation'] and after['selected_configuration'] == before['selected_configuration'] + terminal_password_attempt(password, succeeds=True) + after = status() + assert after['selected_configuration'] is None and after['generation'] == before['generation'] + 1 + assert after['requested_enabled'] == before['requested_enabled'] + if before['selected_configuration']: cli('configuration', 'select', before['selected_configuration']) + finally: + policy.unlink(missing_ok=True) + command('userdel', name) + + class Handler(http.server.BaseHTTPRequestHandler): def do_GET(self): self.send_response(200) @@ -55,7 +170,12 @@ def main(): # an assertion that product install succeeds with the VM's unsupported core. Path('/etc/roentgen').mkdir(mode=0o700, exist_ok=True) Path('/var/lib/roentgen').mkdir(mode=0o700, exist_ok=True) - assets = ROOT / 'assets' + guard = Path('/etc/roentgen/guard.toml') + if not guard.exists(): + assert not Path('/var/lib/roentgen/state.sqlite3').exists(), 'Refusing to fabricate guard input for existing state' + guard.write_text('version = 1\nenabled = false\n') + guard.chmod(0o600) + assets = ROOT / 'assets' destinations = { 'roentgen.service': '/etc/systemd/system/roentgen.service', 'roentgen-guard.service': '/etc/systemd/system/roentgen-guard.service', @@ -97,6 +217,10 @@ def main(): assert status()['selected_configuration'] == selected and status()['generation'] == before['generation'] assert 'authorization' in denied.stderr.lower() or 'accessdenied' in denied.stderr.lower() results.append('noninteractive non-root mutation is denied with durable state unchanged') + cancel_terminal_authorization() + results.append('terminal polkit prompts under an ordinary user; cancellation leaves selection, intent and generation unchanged') + password_authorization_fixture() + results.append('terminal polkit rejects a wrong password and authorizes the real mutation after valid VM-fixture PAM authentication') policy.write_text('polkit.addRule(function(action, subject) { if (action.id == "org.roentgen.control" && subject.user == "rustacean") return polkit.Result.YES; });\n') policy.chmod(0o644) time.sleep(1) -- cgit v1.2.3