1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
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()
|
