Something went wrong. Try again.
Backend environment for match hosting for lance.blue
Something went wrong. Try again.
Python
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129#!/usr/bin/env python3"""Screenshot a headless Chrome page via the DevTools Protocol.
Exists because this box has no node and no python websocket library, and`chrome --screenshot` combined with --virtual-time-budget fires before aWebSocket app (like Suramadu) has finished connecting. A minimal RFC6455 clientis less trouble than the workarounds.
Usage: cdp-screenshot.py <debug-port> <out.png> [url-substring]"""
import base64import jsonimport osimport socketimport structimport sysimport urllib.request
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")
def main(): port = int(sys.argv[1]) out = sys.argv[2] match = sys.argv[3] if len(sys.argv) > 3 else ""
targets = json.load( urllib.request.urlopen(f"http://localhost:{port}/json/list", timeout=15) ) pages = [ t for t in targets if t.get("type") == "page" and match in t.get("url", "") ] if not pages: print(f"no page target matching {match!r}", file=sys.stderr) return 1 ws_url = pages[0]["webSocketDebuggerUrl"] path = ws_url.split(f"{port}", 1)[1]
sock, rest = ws_connect("localhost", port, path) ws_send( sock, json.dumps( {"id": 1, "method": "Page.captureScreenshot", "params": {"format": "png"}} ), ) for text in ws_frames(sock, rest): msg = json.loads(text) if msg.get("id") == 1: if "error" in msg: print(f"CDP error: {msg['error']}", file=sys.stderr) return 1 with open(out, "wb") as fh: fh.write(base64.b64decode(msg["result"]["data"])) print(f"wrote {out} ({os.path.getsize(out)} bytes) from {pages[0]['url']}") return 0 return 1
if __name__ == "__main__": sys.exit(main())