148 lines
4.5 KiB
Python
Executable File
148 lines
4.5 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
import base64
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import socket
|
|
import struct
|
|
import sys
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
|
|
CDP_ENDPOINT = os.environ.get("CDP_ENDPOINT", "http://127.0.0.1:9222")
|
|
|
|
|
|
def get_page_ws_url():
|
|
with urllib.request.urlopen(f"{CDP_ENDPOINT}/json/list", timeout=30) as response:
|
|
targets = json.loads(response.read().decode("utf-8"))
|
|
pages = [target for target in targets if target.get("type") == "page"]
|
|
if not pages:
|
|
raise SystemExit("No CDP page target found.")
|
|
selected = next((page for page in pages if "chatgpt.com/share/" in page.get("url", "")), pages[0])
|
|
return selected["webSocketDebuggerUrl"], selected
|
|
|
|
|
|
def websocket_connect(ws_url):
|
|
parsed = urllib.parse.urlparse(ws_url)
|
|
if parsed.scheme != "ws":
|
|
raise SystemExit(f"Only ws:// endpoints are supported, got {ws_url}")
|
|
host = parsed.hostname or "127.0.0.1"
|
|
port = parsed.port or 80
|
|
path = parsed.path
|
|
if parsed.query:
|
|
path += "?" + parsed.query
|
|
|
|
sock = socket.create_connection((host, port), timeout=30)
|
|
key = base64.b64encode(os.urandom(16)).decode("ascii")
|
|
request = (
|
|
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"
|
|
).encode("ascii")
|
|
sock.sendall(request)
|
|
response = b""
|
|
while b"\r\n\r\n" not in response:
|
|
response += sock.recv(4096)
|
|
header = response.split(b"\r\n\r\n", 1)[0].decode("iso-8859-1")
|
|
if " 101 " not in header.split("\r\n", 1)[0]:
|
|
raise SystemExit(f"WebSocket upgrade failed:\n{header}")
|
|
accept = base64.b64encode(hashlib.sha1((key + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11").encode("ascii")).digest()).decode("ascii")
|
|
if f"Sec-WebSocket-Accept: {accept}".lower() not in header.lower():
|
|
raise SystemExit("WebSocket accept header did not match.")
|
|
return sock
|
|
|
|
|
|
def send_frame(sock, payload):
|
|
data = payload.encode("utf-8")
|
|
header = bytearray([0x81])
|
|
if len(data) < 126:
|
|
header.append(0x80 | len(data))
|
|
elif len(data) < 65536:
|
|
header.append(0x80 | 126)
|
|
header.extend(struct.pack("!H", len(data)))
|
|
else:
|
|
header.append(0x80 | 127)
|
|
header.extend(struct.pack("!Q", len(data)))
|
|
mask = os.urandom(4)
|
|
header.extend(mask)
|
|
masked = bytes(byte ^ mask[index % 4] for index, byte in enumerate(data))
|
|
sock.sendall(bytes(header) + masked)
|
|
|
|
|
|
def recv_exact(sock, size):
|
|
chunks = []
|
|
remaining = size
|
|
while remaining:
|
|
chunk = sock.recv(remaining)
|
|
if not chunk:
|
|
raise EOFError("WebSocket closed")
|
|
chunks.append(chunk)
|
|
remaining -= len(chunk)
|
|
return b"".join(chunks)
|
|
|
|
|
|
def recv_frame(sock):
|
|
first, second = recv_exact(sock, 2)
|
|
opcode = first & 0x0F
|
|
masked = second & 0x80
|
|
length = second & 0x7F
|
|
if length == 126:
|
|
length = struct.unpack("!H", recv_exact(sock, 2))[0]
|
|
elif length == 127:
|
|
length = struct.unpack("!Q", recv_exact(sock, 8))[0]
|
|
mask = recv_exact(sock, 4) if masked else b""
|
|
payload = recv_exact(sock, length) if length else b""
|
|
if masked:
|
|
payload = bytes(byte ^ mask[index % 4] for index, byte in enumerate(payload))
|
|
if opcode == 0x8:
|
|
raise EOFError("WebSocket closed")
|
|
if opcode == 0x9:
|
|
return None
|
|
if opcode != 0x1:
|
|
return None
|
|
return payload.decode("utf-8")
|
|
|
|
|
|
def call(sock, method, params=None, message_id=1):
|
|
send_frame(sock, json.dumps({"id": message_id, "method": method, "params": params or {}}))
|
|
while True:
|
|
message = recv_frame(sock)
|
|
if not message:
|
|
continue
|
|
parsed = json.loads(message)
|
|
if parsed.get("id") == message_id:
|
|
return parsed
|
|
|
|
|
|
def main():
|
|
expression = sys.stdin.read().strip()
|
|
if not expression:
|
|
expression = "({ title: document.title, url: location.href, text: document.body.innerText.slice(0, 1000) })"
|
|
|
|
ws_url, target = get_page_ws_url()
|
|
sock = websocket_connect(ws_url)
|
|
try:
|
|
result = call(
|
|
sock,
|
|
"Runtime.evaluate",
|
|
{
|
|
"expression": expression,
|
|
"returnByValue": True,
|
|
"awaitPromise": True,
|
|
"timeout": 30000,
|
|
},
|
|
)
|
|
finally:
|
|
sock.close()
|
|
|
|
print(json.dumps({"target": target, "result": result}, ensure_ascii=False, indent=2))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|