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
|
#!/usr/bin/env python3
"""Run controller acceptance against real HTTPS and Trojan fixtures on the ALT VM."""
import argparse
import contextlib
import http.server
import json
import os
from pathlib import Path
import socket
import ssl
import subprocess
import tempfile
import threading
import time
import uuid
from email.utils import formatdate
ROOT = Path('/var/tmp/roentgen-implementation')
def run(*args):
return subprocess.run(args, check=True, capture_output=True, text=True, timeout=30)
def available_port():
with socket.socket() as listener:
listener.bind(('127.0.0.1', 0))
return listener.getsockname()[1]
def certificate(directory, ca=False, signed=False):
name = 'ca' if ca else ('server' if signed else 'untrusted')
key, cert = directory / (name + '.key'), directory / (name + '.pem')
if ca:
run('openssl', 'req', '-x509', '-newkey', 'rsa:2048', '-nodes', '-days', '1', '-subj', '/CN=Roentgen VM fixture CA', '-addext', 'basicConstraints=critical,CA:TRUE', '-keyout', str(key), '-out', str(cert))
elif signed:
csr = directory / 'server.csr'
extension = directory / 'server.ext'
extension.write_text('subjectAltName=DNS:localhost,IP:127.0.0.1\nextendedKeyUsage=serverAuth\nbasicConstraints=CA:FALSE\n')
run('openssl', 'req', '-new', '-newkey', 'rsa:2048', '-nodes', '-subj', '/CN=localhost', '-keyout', str(key), '-out', str(csr))
run('openssl', 'x509', '-req', '-in', str(csr), '-CA', str(directory/'ca.pem'), '-CAkey', str(directory/'ca.key'), '-CAcreateserial', '-days', '1', '-extfile', str(extension), '-out', str(cert))
else:
run('openssl', 'req', '-x509', '-newkey', 'rsa:2048', '-nodes', '-days', '1', '-subj', '/CN=localhost', '-addext', 'subjectAltName=DNS:localhost,IP:127.0.0.1', '-keyout', str(key), '-out', str(cert))
return cert, key
def main(binary):
assert os.geteuid() == 0 and socket.gethostname() == 'altctlrs'
assert (ROOT / 'allow-network-tests').exists()
with tempfile.TemporaryDirectory(prefix='roentgen-diagnostics-') as temporary:
directory = Path(temporary)
ca, _ = certificate(directory, ca=True)
cert, key = certificate(directory, signed=True)
bad_cert, bad_key = certificate(directory)
anchor = Path('/etc/pki/ca-trust/source/anchors') / f'roentgen-vm-{uuid.uuid4().hex}.pem'
anchor.write_bytes(ca.read_bytes()); anchor.chmod(0o644)
requests = directory / 'requests.jsonl'; requests.touch()
lock = threading.Lock()
class Handler(http.server.BaseHTTPRequestHandler):
def record(self, method):
with lock:
with requests.open('a') as output:
output.write(json.dumps({'method':method,'path':self.path,'user_agent':self.headers.get('User-Agent'), 'proxy_authorization':self.headers.get('Proxy-Authorization')}) + '\n')
def do_CONNECT(self):
self.record('CONNECT'); self.send_error(502)
def do_GET(self):
self.record('GET')
if self.path == '/rate':
self.send_response(429); self.send_header('Retry-After', '60'); self.end_headers(); return
if self.path == '/rate-date':
self.send_response(503); self.send_header('Retry-After', formatdate(time.time()+65, usegmt=True)); self.end_headers(); return
if self.path == '/redirect':
self.send_response(302); self.send_header('Location', '/empty'); self.end_headers(); return
if self.path == '/partial':
self.send_response(206); self.end_headers(); self.wfile.write(b'partial'); return
self.send_response(200)
if self.path == '/error': self.send_header('Content-Type', 'application/problem+json')
if self.path in ['/json-error', '/long-error', '/late-error', '/escaped-error', '/json-data']: self.send_header('Content-Type', 'application/json')
body = b''
if self.path in ['/error', '/json-error']: body = b'{"error":"fixture-provider-error"}'
if self.path == '/long-error': body = json.dumps({'error':'x'*30000}).encode()
if self.path == '/late-error': body = json.dumps({'metadata':'x'*30000, 'error':'provider failure'}).encode()
if self.path == '/escaped-error': body = b'{"\\u0065rror":"' + b'x'*20000 + b'"}'
if self.path == '/json-data': body = json.dumps({'data':{'error':'nested payload'},'metadata':'x'*30000 + '"error":not a field', 'error':None,'errors':[]}).encode()
if self.path == '/late-html': body = b' ' * 30000 + b'<html>provider error</html>'
if self.path == '/html': body = b'<html>unexpected page</html>'
if self.path == '/oversize': body = b'x' * 4097
if self.path == '/large': body = b'x' * (8 * 1048576)
if self.path != '/slow': self.send_header('Content-Length', str(len(body)))
self.end_headers()
try:
if self.path == '/slow':
self.wfile.write(b'x' * 16384); self.wfile.flush(); time.sleep(3)
else:
self.wfile.write(body)
except (BrokenPipeError, ConnectionResetError, ssl.SSLError): pass
def log_message(self, *args): pass
class Server(http.server.ThreadingHTTPServer):
def handle_error(self, *args): pass
def origin(cert, key):
server = Server(('127.0.0.1', 0), Handler)
context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
context.load_cert_chain(cert, key)
server.socket = context.wrap_socket(server.socket, server_side=True)
threading.Thread(target=server.serve_forever, daemon=True).start()
return server
servers = []
trojan = None
try:
run('update-ca-trust')
servers = [origin(cert,key), origin(bad_cert,bad_key)]
port = available_port()
access = directory / 'trojan-access.log'
config = directory / 'trojan.json'
config.write_text(json.dumps({'log':{'loglevel':'warning','access':str(access)}, 'inbounds':[{'listen':'127.0.0.1','port':port,'protocol':'trojan','settings':{'clients':[{'password':'fixture-password'}]},'streamSettings':{'network':'tcp','security':'tls','tlsSettings':{'certificates':[{'certificateFile':str(cert),'keyFile':str(key)}]}}}], 'routing':{'domainStrategy':'AsIs','rules':[{'type':'field','ip':['127.0.0.1/32'],'outboundTag':'direct'}]}, 'outbounds':[{'tag':'block','protocol':'blackhole','settings':{}},{'tag':'direct','protocol':'freedom','settings':{}}]}))
trojan = subprocess.Popen(['/usr/bin/xray','run','-config',str(config)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
deadline = time.monotonic()+5
while time.monotonic() < deadline:
try:
with socket.create_connection(('127.0.0.1', port), timeout=0.1): break
except OSError: time.sleep(0.05)
else: raise AssertionError('Trojan fixture did not start')
metadata = directory / 'fixture.json'
metadata.write_text(json.dumps({'origin':f'https://127.0.0.1:{servers[0].server_port}', 'bad_origin':f'https://127.0.0.1:{servers[1].server_port}', 'trojan_port':port, 'access_log':str(access), 'requests_log':str(requests)}))
environment = dict(os.environ, ROENTGEN_DIAGNOSTIC_FIXTURE=str(metadata), ROENTGEN_TEST_BINARY=str(ROOT/'roentgen'), SSL_CERT_FILE=str(ca), HTTPS_PROXY='http://127.0.0.1:1', HTTP_PROXY='http://127.0.0.1:1', ALL_PROXY='http://127.0.0.1:1')
subprocess.run([binary, '--nocapture', '--test-threads=1'], env=environment, check=True, timeout=120)
finally:
if trojan:
trojan.terminate()
with contextlib.suppress(subprocess.TimeoutExpired): trojan.wait(timeout=3)
if trojan.poll() is None: trojan.kill(); trojan.wait()
for server in servers: server.shutdown(); server.server_close()
anchor.unlink(missing_ok=True)
run('update-ca-trust')
if __name__ == '__main__':
parser=argparse.ArgumentParser()
parser.add_argument('binary')
main(parser.parse_args().binary)
|
