Implement Linux proxy controller, subscription state, and CLI/TUI

1 files changed, 144 insertions, 0 deletions

diff --git a/tests/vm_controller_acceptance.py b/tests/vm_controller_acceptance.py
new file mode 100644
index 0000000..d39b16f
--- /dev/null
+++ b/tests/vm_controller_acceptance.py
@@ -0,0 +1,144 @@
+#!/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 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, 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')
+        cli('subscription', 'delete', 'Personal')
+        assert status()['selected_configuration'] is None
+        results.append('subscription deletion clears selection without replacement')
+        print(json.dumps({'passed': results}, indent=2))
+    finally:
+        server.shutdown()
+        if policy.exists(): policy.unlink()
+        command('systemctl', 'stop', 'roentgen.service', ok=True)
+
+
+if __name__ == '__main__': main()