1 files changed, 125 insertions, 1 deletions
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)