#!/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()