path: root/tests/vm_controller_acceptance.py

#!/usr/bin/env python3
"""Exercise the actual CLI/system-bus/polkit boundary on the supplied ALT VM."""
import contextlib
import http.server
import json
import os
from pathlib import Path
import socket
import sqlite3
import pty
import select
import termios
import fcntl
import struct
import pwd
import signal
import secrets
import subprocess
import threading
import time

ROOT = Path('/var/tmp/roentgen-implementation')
BIN = str(ROOT / 'roentgen')
BODY = b'trojan://never-print-password@example.com:443?security=tls#First\n'


def command(*args, input=None, ok=True, user=None, env=None):
    argv = (["runuser", "-u", user, "--"] if user else []) + list(args)
    result = subprocess.run(argv, input=input, text=True, capture_output=True, timeout=40, env=env)
    if ok and result.returncode:
        raise AssertionError(f'{args[0:2]} failed: {result.stderr}')
    if not ok and not result.returncode:
        raise AssertionError(f'{args[0:2]} unexpectedly succeeded')
    return result


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)
        self.send_header('Content-Type', 'text/plain')
        self.end_headers()
        self.wfile.write(BODY)
    def log_message(self, *args): pass


def main():
    global BODY
    assert os.geteuid() == 0 and socket.gethostname() == 'altctlrs'
    assert (ROOT / 'allow-network-tests').exists()
    # Test setup installs integration fixtures directly. This is deliberately not
    # 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)
    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',
        'roentgen-xray.service': '/etc/systemd/system/roentgen-xray.service',
        'org.roentgen.Controller1.conf': '/etc/dbus-1/system.d/org.roentgen.Controller1.conf',
        'org.roentgen.policy': '/usr/share/polkit-1/actions/org.roentgen.policy',
    }
    for source, destination in destinations.items():
        Path(destination).write_bytes((assets / source).read_bytes())
        Path(destination).chmod(0o644)
    target = Path('/usr/local/bin/roentgen')
    target.write_bytes(Path(BIN).read_bytes()); target.chmod(0o755)
    command('systemctl', 'daemon-reload')
    command('busctl', 'call', 'org.freedesktop.DBus', '/org/freedesktop/DBus', 'org.freedesktop.DBus', 'ReloadConfig')
    command('systemctl', 'start', 'roentgen-guard.service')
    command('systemctl', 'start', 'roentgen.service')
    server = http.server.ThreadingHTTPServer(('127.0.0.1', 0), Handler)
    threading.Thread(target=server.serve_forever, daemon=True).start()
    token_url = f'http://127.0.0.1:{server.server_port}/never-print-subscription-token'
    if 'Personal' in cli('subscription', 'list').stdout: cli('subscription', 'delete', 'Personal')
    results = []
    policy = Path('/etc/polkit-1/rules.d/00-roentgen-vm-test.rules')
    try:
        cli('subscription', 'add', '--name', 'Personal', input=token_url + '\n')
        before = status()
        assert before['selected_configuration'] is None and not before['requested_enabled']
        public = cli('subscription', 'list').stdout + cli('configuration', 'list').stdout + cli('status', '--json').stdout
        assert 'never-print-password' not in public and 'never-print-subscription-token' not in public
        results.append('real HTTP import remains disabled/unselected and public CLI output omits credentials')
        cli('enable', ok=False)
        assert status()['generation'] == before['generation']
        rows = command('busctl', 'call', 'org.roentgen.Controller1', '/org/roentgen/Controller1', 'org.roentgen.Controller1', 'Configurations', 's', '').stdout
        # busctl prints a quoted JSON string. Decode that string before its JSON payload.
        entries = json.loads(json.loads(rows[2:].strip()))
        selected = entries[0]['id']
        cli('configuration', 'select', selected)
        before = status()
        denied = cli('configuration', 'unset', ok=False, user='rustacean')
        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)
        cli('configuration', 'unset', user='rustacean')
        assert status()['selected_configuration'] is None
        results.append('polkit site allow policy authorizes the actual ordinary bus caller')
        policy.unlink()
        time.sleep(1)
        cli('configuration', 'select', selected)
        BODY = b'trojan://never-print-password@example.com:443#Renamed\nvless://id@example.com:443#Unsupported\ntrojan://broken@host#Invalid\n'
        cli('subscription', 'update', 'Personal')
        listing = cli('configuration', 'list').stdout
        assert 'Renamed' in listing and 'unsupported' in listing and 'invalid' in listing
        assert status()['selected_configuration'] == selected
        results.append('mixed update retains ordered invalid/unsupported rows and normalized identity')
        BODY = b'<html>provider login page</html>'
        cli('subscription', 'update', 'Personal', ok=False)
        assert status()['selected_configuration'] == selected
        BODY = b''
        cli('subscription', 'update', 'Personal')
        assert status()['selected_configuration'] is None
        results.append('whole-document failure preserves selection while received empty list clears it')
        BODY = b'trojan://never-print-password@example.com:443#First\n'
        cli('subscription', 'update', '--all')
        output = command('busctl', 'call', 'org.roentgen.Controller1', '/org/roentgen/Controller1', 'org.roentgen.Controller1', 'Configurations', 's', '').stdout
        selected = json.loads(json.loads(output[2:].strip()))[0]['id']
        cli('configuration', 'select', selected)
        error = cli('enable', ok=False).stderr
        assert 'Unsupported Xray release' in error
        assert not status()['requested_enabled']
        results.append('unsupported packaged Xray activation is rejected before enabled intent changes')
        command('systemctl', 'restart', 'roentgen.service')
        assert status()['selected_configuration'] == selected
        results.append('service restart retains explicit selection and disabled intent')
        editor = ROOT / 'check-editor.py'
        editor.write_text('import os,pathlib,sys\np=pathlib.Path(sys.argv[-1])\nassert p.suffix==".txt" and (p.stat().st_mode & 0o777)==0o600\nassert p.stat().st_uid==os.getuid()\nassert sys.argv[1:3]==["--literal","two words"]\npathlib.Path("/var/tmp/roentgen-editor-path").write_text(str(p))\n')
        env = dict(os.environ, TMPDIR='/var/tmp', EDITOR=f'python3 {editor} --literal "two words"')
        cli('logs', '--editor', user='rustacean', env=env)
        snapshot = Path('/var/tmp/roentgen-editor-path').read_text()
        assert not Path(snapshot).exists()
        results.append('editor receives literal arguments and a user-owned private snapshot that is removed afterward')
        before = status()
        process, terminal = pty.fork()
        if process == 0:
            fcntl.ioctl(0, termios.TIOCSWINSZ, struct.pack('HHHH', 32, 120, 0, 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['TMPDIR'] = '/var/tmp'
            os.execv(BIN, [BIN, 'tui'])
        output = b''
        deadline = time.monotonic() + 10
        while b'Connections' not in output and time.monotonic() < deadline:
            if select.select([terminal], [], [], 0.2)[0]: output += os.read(terminal, 65536)
        assert b'Connections' in output, 'TUI did not render under the invoking ordinary user'
        os.write(terminal, b'l?')
        time.sleep(0.3)
        os.write(terminal, b'\x1b')
        time.sleep(0.1)
        os.write(terminal, b'q')
        done = False
        while time.monotonic() < deadline:
            if select.select([terminal], [], [], 0.1)[0]:
                with contextlib.suppress(OSError): output += os.read(terminal, 65536)
            pid, exit_status = os.waitpid(process, os.WNOHANG)
            if pid:
                assert os.waitstatus_to_exitcode(exit_status) == 0
                done = True
                break
        if not done:
            os.kill(process, 9); os.waitpid(process, 0)
            raise AssertionError('TUI did not exit after q')
        attributes = termios.tcgetattr(terminal)
        assert attributes[3] & termios.ICANON and attributes[3] & termios.ECHO
        assert b'\x1b[?1049l' in output
        os.close(terminal)
        assert status()['selected_configuration'] == before['selected_configuration']
        assert status()['requested_enabled'] == before['requested_enabled']
        results.append('ordinary-user TUI renders, accepts navigation/help/quit, restores terminal modes, and preserves service intent')
        # Persisted enabled intent must not bypass a rejected preflight on retry.
        command('systemctl', 'stop', 'roentgen.service')
        with sqlite3.connect('/var/lib/roentgen/state.sqlite3') as database:
            database.execute('UPDATE state SET enabled=1 WHERE singleton=1')
        Path('/etc/roentgen/guard.toml').write_text('version = 1\nenabled = true\n')
        cli('guard')
        command('systemctl', 'start', 'roentgen.service')
        time.sleep(6)
        current = status()
        assert current['requested_enabled'] and current['routing'] == 'blocked'
        assert 'Unsupported Xray release' in current['last_error']
        assert 'roentgen0' not in command('ip', 'link', 'show').stdout
        results.append('restart and scheduled retry preserve blocked intent after failed compatibility checks')
        # An unset selection must discard stale capture even if a TUN survived.
        command('systemctl', 'stop', 'roentgen.service')
        command('ip', 'tuntap', 'add', 'dev', 'roentgen0', 'mode', 'tun')
        command('ip', 'link', 'set', 'roentgen0', 'up')
        command('ip', 'route', 'add', 'default', 'dev', 'roentgen0', 'table', '20260', 'proto', '186')
        command('ip', 'rule', 'add', 'priority', '10260', 'fwmark', '0x02000000/0x02000000', 'lookup', '20260')
        with sqlite3.connect('/var/lib/roentgen/state.sqlite3') as database:
            database.execute('UPDATE state SET selected=NULL WHERE singleton=1')
        command('systemctl', 'start', 'roentgen.service')
        assert status()['routing'] == 'blocked' and status()['selected_configuration'] is None
        assert '10260' not in command('ip', 'rule', 'show').stdout
        assert 'oifname "roentgen0" accept' not in command('nft', 'list', 'table', 'inet', 'roentgen').stdout
        command('ip', 'link', 'del', 'roentgen0')
        results.append('restart with unset selection removes stale capture and keeps TUN forwarding blocked')
        # Recovery does not depend on parsing a broken editable definition file.
        config = Path('/etc/roentgen/config.toml')
        saved = config.read_bytes()
        config.write_text('broken = [')
        cli('recover', '--disable')
        with sqlite3.connect('/var/lib/roentgen/state.sqlite3') as database:
            assert database.execute('SELECT enabled FROM state').fetchone()[0] == 0
        config.write_bytes(saved)
        command('systemctl', 'start', 'roentgen.service')
        assert not status()['requested_enabled'] and status()['routing'] == 'direct'
        results.append('explicit recovery persists disabled intent even with malformed editable TOML')
        cli('subscription', 'delete', 'Personal')
        assert status()['selected_configuration'] is None
        results.append('subscription deletion clears selection without replacement')
        command('systemctl', 'stop', 'roentgen.service')
        with sqlite3.connect('/var/lib/roentgen/state.sqlite3') as database:
            database.execute('UPDATE state SET enabled=1 WHERE singleton=1')
        Path('/etc/roentgen/guard.toml').write_text('version = 1\nenabled = true\n')
        cli('guard')
        database = Path('/var/lib/roentgen/state.sqlite3')
        database.replace('/var/lib/roentgen/state.sqlite3.missing-fixture')
        command('systemctl', 'start', 'roentgen.service', ok=False)
        assert not database.exists()
        assert 'policy drop' in command('nft', 'list', 'table', 'inet', 'roentgen').stdout
        cli('recover', '--disable')
        with sqlite3.connect(database) as connection:
            assert connection.execute('SELECT enabled FROM state').fetchone()[0] == 0
        results.append('missing durable enabled state fails closed; only explicit recovery initializes disabled state')
        cli('uninstall')
        assert Path('/etc/roentgen/config.toml').exists() and Path('/var/lib/roentgen/state.sqlite3').exists()
        assert not target.exists() and not Path('/etc/systemd/system/roentgen.service').exists()
        assert 'roentgen' not in command('nft', 'list', 'tables').stdout
        # Purge is exercised after re-staging the service assets as a VM fixture.
        for source, destination in destinations.items():
            Path(destination).write_bytes((assets / source).read_bytes())
        target.write_bytes(Path(BIN).read_bytes()); target.chmod(0o755)
        command('systemctl', 'daemon-reload')
        cli('uninstall', '--purge')
        assert not Path('/etc/roentgen').exists() and not Path('/var/lib/roentgen').exists()
        assert command('/usr/bin/xray', 'version').stdout.startswith('Xray 26.3.27')
        results.append('uninstall retains saved data by default, explicit purge removes it, and system Xray remains installed')
        print(json.dumps({'passed': results}, indent=2))
    finally:
        server.shutdown()
        if policy.exists(): policy.unlink()
        subprocess.run(['systemctl', 'stop', 'roentgen.service'], capture_output=True)


if __name__ == '__main__': main()