#!/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 == '/json-error': 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 == '/html': body = b'unexpected page' 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)