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 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)


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)
    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')
        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()