Something went wrong. Try again.
Backend environment for match hosting for lance.blue
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178#!/usr/bin/env python3"""A minimal CDP client over a hand-rolled RFC6455 WebSocket.
Shared by scripts/perf/latency-pilot.py and scripts/perf/startup.py, which bothneed to drive a page in an already-running Chrome and read values out of it.This box has no node and no python websocket library, and one is not worthadding for two scripts - tests/screenshot.py carries the same client for thesame reason."""
import base64import itertoolsimport jsonimport osimport socketimport structimport timeimport urllib.request
# --- minimal CDP client ------------------------------------------------------# Same RFC6455 client as tests/screenshot.py: this box has no node and no# python websocket library.
def ws_connect(host, port, path): sock = socket.create_connection((host, port), timeout=30) key = base64.b64encode(os.urandom(16)).decode() req = ( f"GET {path} HTTP/1.1\r\n" f"Host: {host}:{port}\r\n" "Upgrade: websocket\r\n" "Connection: Upgrade\r\n" f"Sec-WebSocket-Key: {key}\r\n" "Sec-WebSocket-Version: 13\r\n\r\n" ) sock.sendall(req.encode()) buf = b"" while b"\r\n\r\n" not in buf: chunk = sock.recv(4096) if not chunk: raise RuntimeError("handshake closed early") buf += chunk if b"101" not in buf.split(b"\r\n")[0]: raise RuntimeError(f"upgrade failed: {buf.split(b'\r\n')[0]!r}") return sock, buf.split(b"\r\n\r\n", 1)[1]
def ws_send(sock, payload): data = payload.encode() header = bytearray([0x81]) # FIN + text mask = os.urandom(4) n = len(data) if n < 126: header.append(0x80 | n) elif n < (1 << 16): header.append(0x80 | 126) header += struct.pack(">H", n) else: header.append(0x80 | 127) header += struct.pack(">Q", n) header += mask sock.sendall(bytes(header) + bytes(b ^ mask[i % 4] for i, b in enumerate(data)))
def ws_frames(sock, initial=b""): """Yield decoded text payloads. Server->client frames are never masked.""" buf = bytearray(initial)
def need(n): while len(buf) < n: chunk = sock.recv(65536) if not chunk: raise RuntimeError("connection closed") buf.extend(chunk)
while True: need(2) length = buf[1] & 0x7F offset = 2 if length == 126: need(4) length = struct.unpack(">H", bytes(buf[2:4]))[0] offset = 4 elif length == 127: need(10) length = struct.unpack(">Q", bytes(buf[2:10]))[0] offset = 10 need(offset + length) payload = bytes(buf[offset : offset + length]) del buf[: offset + length] yield payload.decode("utf-8", "replace")
class CDP: """One page's debugger connection: send a command, wait for its reply.
Frames that are not the reply we are waiting on are events (or replies to calls nobody is blocking on); they are queued in `events` rather than dropped, so a caller that wants Page.loadEventFired can look for it. """
def __init__(self, host, port, ws_path): self.sock, rest = ws_connect(host, port, ws_path) self._frames = ws_frames(self.sock, rest) self._ids = itertools.count(1) self.events = []
def send(self, method, params=None): msg_id = next(self._ids) ws_send( self.sock, json.dumps({"id": msg_id, "method": method, "params": params or {}}), ) return msg_id
def call(self, method, params=None, timeout=30): msg_id = self.send(method, params) deadline = time.monotonic() + timeout while time.monotonic() < deadline: self.sock.settimeout(max(0.1, deadline - time.monotonic())) try: text = next(self._frames) except TimeoutError: break msg = json.loads(text) if msg.get("id") == msg_id: if "error" in msg: raise RuntimeError(f"{method} failed: {msg['error']}") return msg.get("result", {}) self.events.append(msg) raise RuntimeError(f"{method} timed out after {timeout}s")
def wait_for_event(self, name, timeout=30): deadline = time.monotonic() + timeout for msg in self.events: if msg.get("method") == name: self.events.remove(msg) return msg while time.monotonic() < deadline: self.sock.settimeout(max(0.1, deadline - time.monotonic())) try: text = next(self._frames) except TimeoutError: return None msg = json.loads(text) if msg.get("method") == name: return msg self.events.append(msg) return None
def evaluate(self, expression, timeout=30): result = self.call( "Runtime.evaluate", {"expression": expression, "returnByValue": True, "awaitPromise": True}, timeout=timeout, ) exc = result.get("exceptionDetails") if exc: raise RuntimeError(f"page threw: {exc}") return result.get("result", {}).get("value")
def find_page(host, port, match): targets = json.load( urllib.request.urlopen(f"http://{host}:{port}/json/list", timeout=15) ) pages = [t for t in targets if t.get("type") == "page"] if match: pages = [t for t in pages if match in t.get("url", "")] else: # Prefer a page that has actually navigated somewhere over a blank tab. pages = [ p for p in pages if p.get("url", "") not in ("", "about:blank") ] or pages if not pages: raise RuntimeError(f"no page target matching {match!r}") return pages[0]