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
import time
import zlib

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())
counts = {}
counts_lock = threading.Lock()


def count(token, path):
    with counts_lock:
        key = (token, path)
        counts[key] = counts.get(key, 0) + 1
        return counts[key]


def png(size):
    def chunk(kind, data):
        return struct.pack("!I", len(data)) + kind + data + struct.pack("!I", zlib.crc32(kind + data))
    return (b"\x89PNG\r\n\x1a\n" + chunk(b"IHDR", struct.pack("!IIBBBBB", size, size, 8, 2, 0, 0, 0))
            + chunk(b"IDAT", zlib.compress((b"\x00" + b"\x00\x80\xff" * size) * size)) + chunk(b"IEND", b""))


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

    def reply(self, status, body):
        payload = json.dumps(body).encode()
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(payload)))
        self.end_headers()
        self.wfile.write(payload)

    def snapshot(self, token):
        result = dict(state)
        if token.startswith("live-"):
            with counts_lock:
                commands = counts.get((token, "command"), 0)
            result["title"] = "Track " + str(commands)
            if token == "live-unsupported":
                result["capabilities"] = {key: False for key in result["capabilities"]}
        result["artworkId"] = "A" * 64 if token in ("artwork", "artwork-fail", "late-art") else None
        return result

    def do_POST(self):
        body = json.loads(self.rfile.read(int(self.headers.get("Content-Length", "0"))))
        token = self.headers.get("Authorization", "").removeprefix("Bearer ")
        if token.startswith("live-"):
            count(token, "command")
            if token == "live-reject":
                self.reply(200, {"success": False, "code": "unsupported", "message": "Rejected"})
            else:
                self.reply(200, {"success": True, "code": "ok", "message": "OK"})
            return
        if self.path == "/v1/pair":
            code = body.get("code")
            if code == "33333333":
                time.sleep(0.5)
                self.reply(200, {"success": True, "code": "ok", "token": "A" * 64})
            else:
                error = "pairing_closed" if code == "22222222" else "invalid_code"
                self.reply(403, {"success": False, "code": error})
        else:
            self.reply(200, {"success": False, "code": "unsupported", "message": "Плеер не поддерживает эту команду."})

    def do_GET(self):
        token = self.headers.get("Authorization", "").removeprefix("Bearer ")
        number = count(token, self.path)
        if token == "live-refresh-fail" and self.path == "/v1/state":
            with counts_lock:
                commands = counts.get((token, "command"), 0)
            if commands:
                self.reply(503, {})
                return
        if token == "revoked":
            self.reply(401, {})
            return
        if token == "redirect" and self.path == "/v1/state":
            self.send_response(302)
            self.send_header("Location", "/redirect-target")
            self.send_header("Content-Length", "0")
            self.end_headers()
            return
        if token == "recover" and self.path == "/v1/state" and number == 1:
            self.reply(503, {})
            return
        if self.path.startswith("/v1/artwork/"):
            if token == "artwork-fail" or (token == "artwork" and number == 1):
                self.reply(404, {})
                return
            size = 1
            if token == "late-art":
                if self.path.endswith("A" * 64): time.sleep(0.5)
                else: size = 2
            payload = png(size)
            self.send_response(200)
            self.send_header("Content-Type", "image/png")
            self.send_header("Content-Length", str(len(payload)))
            self.end_headers()
            self.wfile.write(payload)
            return
        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()
            snapshot = self.snapshot(token)
            if token == "late-art":
                time.sleep(0.1)
                snapshot["artworkId"] = "B" * 64
            for _ in range(30 if token == "command-reject" else 1):
                payload = json.dumps({"type": "state", "state": snapshot}).encode()
                self.wfile.write(b"\x81\x7e" + struct.pack("!H", len(payload)) + payload)
                self.wfile.flush()
                if token == "command-reject": time.sleep(0.05)
        else:
            self.reply(200, self.snapshot(token))


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()