224 lines
11 KiB
Python
224 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""Temporary X11 local-cursor proof, NOT a production Moonlight cursor protocol.
|
|
|
|
Requires an already authenticated Moonlight session. Never changes Sunshine.
|
|
The caller owns remote-video-cursor handoff and must restore it in finally.
|
|
Only supports exported alpha cursors without destination-dependent XOR pixels.
|
|
"""
|
|
import argparse
|
|
import ctypes as C
|
|
import fcntl
|
|
import json
|
|
import math
|
|
import os
|
|
from pathlib import Path
|
|
import signal
|
|
import subprocess
|
|
import time
|
|
|
|
|
|
class XcursorImage(C.Structure):
|
|
_fields_ = [("version", C.c_uint32), ("size", C.c_uint32),
|
|
("width", C.c_uint32), ("height", C.c_uint32),
|
|
("xhot", C.c_uint32), ("yhot", C.c_uint32),
|
|
("delay", C.c_uint32), ("pixels", C.POINTER(C.c_uint32))]
|
|
|
|
|
|
class XfixesImage(C.Structure):
|
|
_fields_ = [("x", C.c_short), ("y", C.c_short),
|
|
("width", C.c_ushort), ("height", C.c_ushort),
|
|
("xhot", C.c_ushort), ("yhot", C.c_ushort),
|
|
("serial", C.c_ulong), ("pixels", C.POINTER(C.c_ulong)),
|
|
("atom", C.c_ulong), ("name", C.c_char_p)]
|
|
|
|
|
|
def validate_assets(data):
|
|
if not isinstance(data, dict) or type(data.get("version")) is not int or data.get("version") != 1 or not isinstance(data.get("shapes"), list) or not 1 <= len(data["shapes"]) <= 32:
|
|
raise ValueError("Invalid asset version/count")
|
|
result = {}
|
|
for s in data["shapes"]:
|
|
if not isinstance(s, dict):
|
|
raise ValueError("Invalid shape")
|
|
name = s.get("name")
|
|
if not isinstance(name, str) or not name.isascii() or not name.isalnum() or len(name) > 32 or name in result:
|
|
raise ValueError("Invalid/duplicate shape name")
|
|
for key in ("width", "height", "hotX", "hotY", "invertedPixels"):
|
|
if type(s.get(key)) is not int:
|
|
raise ValueError("Noninteger cursor metadata")
|
|
w, h = s["width"], s["height"]
|
|
if not 1 <= w <= 128 or not 1 <= h <= 128 or not 0 <= s["hotX"] < w or not 0 <= s["hotY"] < h:
|
|
raise ValueError("Invalid size/hotspot")
|
|
if s["invertedPixels"] != 0:
|
|
raise ValueError("Destination-dependent XOR cursor requires fallback")
|
|
pixels = s.get("argb")
|
|
if not isinstance(pixels, list) or len(pixels) != w*h or any(type(p) is not int or not 0 <= p <= 0xffffffff for p in pixels):
|
|
raise ValueError("Invalid ARGB pixels")
|
|
result[name] = s
|
|
return result
|
|
|
|
|
|
def validate_state(state, shapes):
|
|
if not isinstance(state, dict) or type(state.get("version")) is not int or state.get("version") != 1 or type(state.get("visible")) is not bool or not isinstance(state.get("shape"), str) or state.get("shape") not in shapes:
|
|
raise ValueError("Invalid state/unknown shape")
|
|
# A bounded wall-clock lease prevents stale local pointers after the bridge dies.
|
|
expires = state.get("expires")
|
|
if type(expires) not in (int, float) or not math.isfinite(expires) or not time.time() < expires <= time.time()+5:
|
|
raise ValueError("Expired/invalid state lease")
|
|
return state["shape"] if state["visible"] else "__hidden"
|
|
|
|
|
|
def read_json(path, limit):
|
|
with open(path, "rb") as f:
|
|
raw = f.read(limit+1)
|
|
if len(raw) > limit:
|
|
raise ValueError("Oversize JSON")
|
|
return json.loads(raw)
|
|
|
|
|
|
class LocalCursor:
|
|
def __init__(self, window):
|
|
self.x = C.CDLL("libX11.so.6")
|
|
self.c = C.CDLL("libXcursor.so.1")
|
|
self.f = C.CDLL("libXfixes.so.3")
|
|
self.x.XOpenDisplay.argtypes = [C.c_char_p]; self.x.XOpenDisplay.restype = C.c_void_p
|
|
self.x.XDefineCursor.argtypes = [C.c_void_p, C.c_ulong, C.c_ulong]
|
|
self.x.XSync.argtypes = [C.c_void_p, C.c_int]
|
|
self.x.XFree.argtypes = [C.c_void_p]
|
|
self.x.XFreeCursor.argtypes = [C.c_void_p, C.c_ulong]
|
|
self.x.XCloseDisplay.argtypes = [C.c_void_p]
|
|
self.f.XFixesGetCursorImage.argtypes = [C.c_void_p]; self.f.XFixesGetCursorImage.restype = C.POINTER(XfixesImage)
|
|
self.f.XFixesHideCursor.argtypes = [C.c_void_p, C.c_ulong]
|
|
self.f.XFixesShowCursor.argtypes = [C.c_void_p, C.c_ulong]
|
|
self.c.XcursorImageCreate.argtypes = [C.c_int, C.c_int]; self.c.XcursorImageCreate.restype = C.POINTER(XcursorImage)
|
|
self.c.XcursorImageDestroy.argtypes = [C.POINTER(XcursorImage)]
|
|
self.c.XcursorImageLoadCursor.argtypes = [C.c_void_p, C.POINTER(XcursorImage)]; self.c.XcursorImageLoadCursor.restype = C.c_ulong
|
|
self.d = self.x.XOpenDisplay(None)
|
|
if not self.d:
|
|
raise RuntimeError("Cannot open X display")
|
|
self.window = window
|
|
self.cursors = {}
|
|
self.original = self.create(self.snapshot(), premultiplied=True)
|
|
|
|
def snapshot(self):
|
|
image = self.f.XFixesGetCursorImage(self.d)
|
|
if not image:
|
|
raise RuntimeError("No XFixes cursor image")
|
|
try:
|
|
i = image.contents
|
|
if not 1 <= i.width <= 128 or not 1 <= i.height <= 128:
|
|
raise RuntimeError("Unsupported X cursor dimensions")
|
|
return dict(width=i.width, height=i.height, hotX=i.xhot, hotY=i.yhot,
|
|
x=i.x, y=i.y, argb=[int(i.pixels[p]) & 0xffffffff for p in range(i.width*i.height)])
|
|
finally:
|
|
self.x.XFree(image)
|
|
|
|
def create(self, s, premultiplied=False):
|
|
image = self.c.XcursorImageCreate(s["width"], s["height"])
|
|
if not image:
|
|
raise RuntimeError("Xcursor allocation failed")
|
|
try:
|
|
image.contents.xhot, image.contents.yhot = s["hotX"], s["hotY"]
|
|
for j, p in enumerate(s["argb"]):
|
|
if premultiplied:
|
|
image.contents.pixels[j] = p
|
|
continue
|
|
# XRender/Xcursor uses premultiplied alpha; Windows export is straight ARGB.
|
|
a = p >> 24
|
|
r, g, b = ((p >> shift) & 255 for shift in (16, 8, 0))
|
|
image.contents.pixels[j] = (a << 24) | ((r*a+127)//255 << 16) | ((g*a+127)//255 << 8) | ((b*a+127)//255)
|
|
cursor = self.c.XcursorImageLoadCursor(self.d, image)
|
|
if not cursor:
|
|
raise RuntimeError("Xcursor creation failed")
|
|
return cursor
|
|
finally:
|
|
self.c.XcursorImageDestroy(image)
|
|
|
|
def apply(self, cursor):
|
|
self.x.XDefineCursor(self.d, self.window, cursor)
|
|
self.x.XSync(self.d, 0)
|
|
|
|
def close(self, sdl_pid=None):
|
|
self.apply(self.original)
|
|
try:
|
|
if sdl_pid is not None:
|
|
# X11 cursor resources owned by this connection disappear when
|
|
# it closes. Have SDL restore its own hidden cursor first. The
|
|
# temporary XFixes hide prevents the first toggle flashing the
|
|
# default local arrow. This POC requires the same focused stream.
|
|
active = subprocess.check_output(["xdotool", "getwindowfocus", "getwindowpid"],text=True).strip()
|
|
if active != str(sdl_pid):
|
|
raise RuntimeError("Cannot return cursor ownership: stream lost focus; reconnect normal client")
|
|
self.f.XFixesHideCursor(self.d,self.window); self.x.XSync(self.d,0)
|
|
try:
|
|
for unused in range(2):
|
|
subprocess.run(["xdotool","key","ctrl+alt+shift+c"],check=True)
|
|
time.sleep(.08)
|
|
finally:
|
|
self.f.XFixesShowCursor(self.d,self.window); self.x.XSync(self.d,0)
|
|
finally:
|
|
for cursor in [self.original, *self.cursors.values()]:
|
|
self.x.XFreeCursor(self.d, cursor)
|
|
self.x.XCloseDisplay(self.d)
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--window", type=int, required=True)
|
|
parser.add_argument("--pid", type=int, required=True)
|
|
parser.add_argument("--assets", type=Path, required=True)
|
|
parser.add_argument("--state", type=Path, required=True)
|
|
parser.add_argument("--seconds", type=int, default=120)
|
|
parser.add_argument("--snapshot", type=Path)
|
|
args = parser.parse_args()
|
|
if not 1 <= args.seconds <= 180 or args.window <= 0 or args.pid <= 1:
|
|
parser.error("Invalid window/pid/duration")
|
|
command = Path(f"/proc/{args.pid}/cmdline").read_bytes().replace(b"\0", b" ")
|
|
if b"/moonlight " not in command or b" stream " not in command or b" --absolute-mouse " not in command:
|
|
parser.error("Target must be the absolute-mode Moonlight stream")
|
|
actual_pid = subprocess.check_output(["xdotool", "getwindowpid", str(args.window)], text=True).strip()
|
|
focused_window = subprocess.check_output(["xdotool", "getwindowfocus"], text=True).strip()
|
|
if actual_pid != str(args.pid) or focused_window != str(args.window):
|
|
parser.error("Target must be the same focused Moonlight window")
|
|
# Keep the experiment exclusive, with owner-only local files.
|
|
lock = open(args.state.with_suffix(".lock"), "w")
|
|
os.chmod(lock.name, 0o600)
|
|
fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
shapes = validate_assets(read_json(args.assets, 4*1024*1024))
|
|
cursor = LocalCursor(args.window)
|
|
if any(cursor.snapshot()["argb"]):
|
|
cursor.x.XFreeCursor(cursor.d, cursor.original)
|
|
cursor.x.XCloseDisplay(cursor.d)
|
|
parser.error("POC requires the original local cursor to be hidden")
|
|
stop = False
|
|
def stopped(*unused):
|
|
nonlocal stop
|
|
stop = True
|
|
signal.signal(signal.SIGTERM, stopped); signal.signal(signal.SIGINT, stopped)
|
|
try:
|
|
cursor.cursors = {name: cursor.create(s) for name, s in shapes.items()}
|
|
cursor.cursors["__hidden"] = cursor.create(dict(width=1, height=1, hotX=0, hotY=0, argb=[0]))
|
|
current = None
|
|
end = time.monotonic()+args.seconds
|
|
while not stop and time.monotonic() < end:
|
|
if not Path(f"/proc/{args.pid}").exists():
|
|
break
|
|
try:
|
|
desired = validate_state(read_json(args.state, 1024), shapes)
|
|
except (ValueError, OSError) as e:
|
|
print(json.dumps(dict(event="lease-ended", reason=str(e))), flush=True)
|
|
break
|
|
if desired != current:
|
|
cursor.apply(cursor.cursors[desired]); current = desired
|
|
s = cursor.snapshot()
|
|
if args.snapshot:
|
|
args.snapshot.write_text(json.dumps(s))
|
|
print(json.dumps(dict(event="applied", shape=current, x=s["x"], y=s["y"], width=s["width"], height=s["height"], hotX=s["hotX"], hotY=s["hotY"])), flush=True)
|
|
time.sleep(.01)
|
|
finally:
|
|
cursor.close(sdl_pid=args.pid)
|
|
print('{"event":"sdl-hidden-cursor-restored"}', flush=True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|