91 lines
3.6 KiB
Python
91 lines
3.6 KiB
Python
"""Drives headless Chrome over the DevTools protocol with a stdlib-only WebSocket client, to run real WebAuthn ceremonies."""
|
|
import base64, json, os, socket, struct, sys, time, urllib.request
|
|
|
|
class CDP:
|
|
def __init__(self, url):
|
|
host, rest = url[len("ws://"):].split("/", 1)
|
|
name, port = host.split(":")
|
|
self.sock = socket.create_connection((name, int(port)))
|
|
key = base64.b64encode(os.urandom(16)).decode()
|
|
self.sock.sendall((f"GET /{rest} HTTP/1.1\r\nHost: {host}\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n"
|
|
f"Sec-WebSocket-Key: {key}\r\nSec-WebSocket-Version: 13\r\n\r\n").encode())
|
|
response = b""
|
|
while b"\r\n\r\n" not in response:
|
|
response += self.sock.recv(4096)
|
|
assert b" 101 " in response.split(b"\r\n")[0], response
|
|
self.buffer = response.split(b"\r\n\r\n", 1)[1]
|
|
self.next_id = 0
|
|
|
|
def _read(self, count):
|
|
while len(self.buffer) < count:
|
|
chunk = self.sock.recv(65536)
|
|
if not chunk:
|
|
raise ConnectionError("closed")
|
|
self.buffer += chunk
|
|
data, self.buffer = self.buffer[:count], self.buffer[count:]
|
|
return data
|
|
|
|
def _frame(self):
|
|
first, second = self._read(2)
|
|
length = second & 0x7F
|
|
if length == 126:
|
|
length = struct.unpack(">H", self._read(2))[0]
|
|
elif length == 127:
|
|
length = struct.unpack(">Q", self._read(8))[0]
|
|
return first & 0x0F, first & 0x80, self._read(length)
|
|
|
|
def _message(self):
|
|
payload = b""
|
|
while True:
|
|
opcode, final, data = self._frame()
|
|
if opcode in (0x0, 0x1, 0x2):
|
|
payload += data
|
|
if final:
|
|
return json.loads(payload)
|
|
|
|
def send(self, method, **params):
|
|
self.next_id += 1
|
|
data = json.dumps({"id": self.next_id, "method": method, "params": params}).encode()
|
|
mask = os.urandom(4)
|
|
header = bytes([0x81])
|
|
if len(data) < 126:
|
|
header += bytes([0x80 | len(data)])
|
|
elif len(data) < 65536:
|
|
header += bytes([0x80 | 126]) + struct.pack(">H", len(data))
|
|
else:
|
|
header += bytes([0x80 | 127]) + struct.pack(">Q", len(data))
|
|
self.sock.sendall(header + mask + bytes(b ^ mask[i % 4] for i, b in enumerate(data)))
|
|
while True:
|
|
message = self._message()
|
|
if message.get("id") == self.next_id:
|
|
if "error" in message:
|
|
raise RuntimeError(f"{method}: {message['error']}")
|
|
return message.get("result", {})
|
|
|
|
def evaluate(self, expression):
|
|
result = self.send("Runtime.evaluate", expression=expression, awaitPromise=True, returnByValue=True)
|
|
if "exceptionDetails" in result:
|
|
raise RuntimeError(result["exceptionDetails"].get("exception", {}).get("description", str(result["exceptionDetails"])))
|
|
return result["result"].get("value")
|
|
|
|
def wait_for(self, expression, seconds=10):
|
|
deadline = time.time() + seconds
|
|
while time.time() < deadline:
|
|
try:
|
|
if self.evaluate(expression):
|
|
return True
|
|
except RuntimeError:
|
|
pass
|
|
time.sleep(0.2)
|
|
return False
|
|
|
|
def connect(port):
|
|
for _ in range(50):
|
|
try:
|
|
targets = json.load(urllib.request.urlopen(f"http://127.0.0.1:{port}/json"))
|
|
page = next(t for t in targets if t["type"] == "page")
|
|
return CDP(page["webSocketDebuggerUrl"])
|
|
except Exception:
|
|
time.sleep(0.2)
|
|
sys.exit("Chrome's DevTools endpoint never answered")
|