update_tls_test.py

1"""Both update requests must reject an untrusted TLS peer before trusting its metadata or bytes."""2import http.server3import json4import os5from pathlib import Path6import ssl7import subprocess8import sys9import tempfile10import threading1112binary = Path(sys.argv[1]).resolve()13work = Path(tempfile.mkdtemp(prefix='koh-update-tls-'))14subprocess.run(['openssl', 'req', '-x509', '-newkey', 'rsa:2048', '-nodes', '-days', '1', '-subj', '/CN=localhost', '-addext', 'subjectAltName=IP:127.0.0.1', '-keyout', str(work / 'key.pem'), '-out', str(work / 'cert.pem')], check=True, capture_output=True)15platforms = ['macos-aarch64', 'macos-x86_64', 'linux-aarch64', 'linux-x86_64']1617class Handler(http.server.BaseHTTPRequestHandler):18    def do_GET(self):19        body = json.dumps({'version': '999.0.0', 'platforms': {p: {'url': tls_url, 'blake3': '0' * 64} for p in platforms}}).encode()20        self.send_response(200)21        self.send_header('Content-Length', str(len(body)))22        self.end_headers()23        self.wfile.write(body)24    def log_message(self, *_):25        pass2627secure = http.server.ThreadingHTTPServer(('127.0.0.1', 0), Handler)28context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)29context.load_cert_chain(work / 'cert.pem', work / 'key.pem')30secure.socket = context.wrap_socket(secure.socket, server_side=True)31tls_url = f'https://127.0.0.1:{secure.server_port}/release'32plain = http.server.ThreadingHTTPServer(('127.0.0.1', 0), Handler)33for server in (secure, plain):34    threading.Thread(target=server.serve_forever, daemon=True).start()35try:36    trusted = dict(os.environ, HOME=str(work), KOH_UPDATE_SOURCE=tls_url, SSL_CERT_FILE=str(work / 'cert.pem'))37    good = subprocess.run([str(binary), 'update', '--check', '--json'], env=trusted, capture_output=True, text=True, timeout=30)38    assert good.returncode == 1 and json.loads(good.stdout)['latest'] == '999.0.0', (good.stdout, good.stderr)39    for url, args, message in [(tls_url, ['--check', '--json'], 'could not reach'), (f'http://127.0.0.1:{plain.server_port}/release', ['--yes', '--json'], 'could not download')]:40        env = dict(os.environ, HOME=str(work), KOH_UPDATE_SOURCE=url)41        result = subprocess.run([str(binary), 'update', *args], env=env, capture_output=True, text=True, timeout=30)42        assert result.returncode == 1 and message in result.stderr, (result.stdout, result.stderr)43    for args in [['remote', f'https://127.0.0.1:{secure.server_port}', '--code', 'x', '--json'], ['steal', f'https://127.0.0.1:{secure.server_port}/nina', '--json'], ['remote', 's3', tls_url, 'test', 'test', '--json']]:44        isolated = dict(os.environ, HOME=str(work), KOH_ACCESS_KEY='test', KOH_SECRET_KEY='test')45        isolated.pop('SSL_CERT_FILE', None)46        result = subprocess.run([str(binary), *args], cwd=work, env=isolated, capture_output=True, text=True, timeout=45)47        assert result.returncode == 1 and 'handshake' in result.stderr, (args, result.stdout, result.stderr)48    print('PASS: trusted TLS accepted; metadata, binaries, connect, steal and S3 reject self-signed TLS')49finally:50    secure.shutdown()51    plain.shutdown()