ios/tools/tls_fixture.py

"""CI-only HTTPS/WSS server with the same certificate shape as the Windows agent."""
import base64
import hashlib
import http.server
import ipaddress
import json
import pathlib
import socket
import socketserver
import ssl
import struct
import subprocess
import threading

root = pathlib.Path(__file__).resolve().parents[1]
work = root / "build/tls"
work.mkdir(parents=True, exist_ok=True)
# Discover the runner's LAN address without sending a packet.
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as probe:
    probe.connect(("192.0.2.1", 80))
    address = probe.getsockname()[0]
assert any(ipaddress.ip_address(address) in ipaddress.ip_network(n)
           for n in ("10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16")), address
config = work / "cert.cnf"
config.write_text("""[req]
distinguished_name = dn
prompt = no
[dn]
CN = MusicBridge Agent
[ext]
basicConstraints = critical,CA:FALSE
keyUsage = critical,digitalSignature,keyEncipherment
extendedKeyUsage = serverAuth
subjectAltName = DNS:localhost,IP:127.0.0.1
""")
state = json.loads((root / "MusicBridgeTests/state-v1.json").read_text())


class Handler(http.server.BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"

    def do_GET(self):
        if self.path == "/v1/events":
            accept = base64.b64encode(hashlib.sha1(
                (self.headers["Sec-WebSocket-Key"] + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11").encode()).digest()).decode()
            self.send_response(101)
            self.send_header("Upgrade", "websocket")
            self.send_header("Connection", "Upgrade")
            self.send_header("Sec-WebSocket-Accept", accept)
            self.end_headers()
            payload = json.dumps({"type": "state", "state": state}).encode()
            self.wfile.write(b"\x81\x7e" + struct.pack("!H", len(payload)) + payload)
            self.wfile.flush()
        else:
            payload = json.dumps(state).encode()
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(payload)))
            self.end_headers()
            self.wfile.write(payload)


fixtures = {}
class Server(http.server.ThreadingHTTPServer):
    def server_bind(self):
        # HTTPServer's reverse DNS lookup can stall on hosted runner addresses.
        socketserver.TCPServer.server_bind(self)
        self.server_name, self.server_port = self.server_address


for name, days in (("valid", "1825"), ("expired", "-1")):
    print(f"Generating {name} certificate for {address}", flush=True)
    key, csr, cert = (work / (name + suffix) for suffix in (".key", ".csr", ".pem"))
    subprocess.run(["openssl", "req", "-new", "-newkey", "rsa:2048", "-nodes", "-config", str(config),
                    "-keyout", str(key), "-out", str(csr)], check=True, capture_output=True, timeout=15)
    subprocess.run(["openssl", "x509", "-req", "-in", str(csr), "-signkey", str(key), "-sha256",
                    "-days", days, "-extfile", str(config), "-extensions", "ext", "-out", str(cert)],
                   check=True, capture_output=True, timeout=15)
    context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    context.minimum_version = ssl.TLSVersion.TLSv1_2
    context.load_cert_chain(cert, key)
    server = Server((address, 0), Handler)
    server.socket = context.wrap_socket(server.socket, server_side=True)
    threading.Thread(target=server.serve_forever, daemon=True).start()
    fixtures[name] = {"endpoint": f"https://{address}:{server.server_port}",
                      "fingerprint": hashlib.sha256(ssl.PEM_cert_to_DER_cert(cert.read_text())).hexdigest()}
(root / "MusicBridgeTests/tls-fixture.json").write_text(json.dumps(fixtures))
print("TLS fixtures ready", flush=True)
threading.Event().wait()