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