#!/usr/bin/python3 """Disposable, separate logind/DRM session on an unused VT. Never logs GNOME out. All system services are transient; an independent timer returns to the original session even if this validation compositor fails. Requires existing sudo access. """ import argparse import importlib.util import json import os from pathlib import Path import pwd import socket import struct import subprocess import time ROOT = Path(__file__).resolve().parents[1] DATA = ROOT / ".artifacts/rime-drm-validation-20261028" UID = ROOT.stat().st_uid USER = pwd.getpwuid(UID).pw_name RUNTIME = Path(f"/run/user/{UID}/ferese-rime-drm-validation") MAIN = "ferese-drm-validation-20261028.service" RETURN = "ferese-drm-return-20261028" IME = "ferese-rime-drm-20261028.service" APP = "ferese-rime-drm-entry-20261028.service" LOCK = "ferese-drm-real-lock-20261028.service" def run(args, **kwargs): return subprocess.check_output(args, text=True, **kwargs).strip() def manager_env(): return {**os.environ, "XDG_RUNTIME_DIR":f"/run/user/{UID}", "DBUS_SESSION_BUS_ADDRESS":f"unix:path=/run/user/{UID}/bus"} def start(): return_seconds=int(os.environ.get("FERESE_DRM_RETURN_SECONDS","60")) if not 30 <= return_seconds <= 180:raise ValueError("Return interval must be 30-180 seconds") DATA.mkdir(mode=0o700,parents=True,exist_ok=True) DATA.chmod(0o700) # Build the disposable protocol fixture from source, never a PAM mock. xml=Path(run(["pkg-config","--variable=pkgdatadir","wayland-protocols"])) / "staging/ext-session-lock/ext-session-lock-v1.xml" for mode,name in (("client-header","ext-session-lock-client-protocol.h"),("private-code","ext-session-lock-protocol.c")): subprocess.run(["wayland-scanner",mode,str(xml),str(DATA/name)],check=True) flags=run(["pkg-config","--cflags","--libs","wayland-client"]).split() subprocess.run(["cc","-Wall","-Wextra","-Werror",f"-I{DATA}",str(ROOT/"scripts/tests/fixtures/session-lock-cycle.c"), str(DATA/"ext-session-lock-protocol.c"),"-o",str(DATA/"lock-cycle"),*flags],check=True) RUNTIME.mkdir(mode=0o700, exist_ok=True) if RUNTIME.stat().st_uid != UID or RUNTIME.stat().st_mode & 0o077: raise RuntimeError("Unexpected private DRM runtime owner/permissions") active = run(["loginctl", "show-seat", "seat0", "-p", "ActiveSession", "--value"]) if not active or run(["loginctl", "show-session", active, "-p", "Name", "--value"]) != USER: raise RuntimeError("The original seat must belong to the repository owner") vt = "3" if any(run(["loginctl", "show-session", sid, "-p", "VTNr", "--value"]) == vt for sid in run(["loginctl", "list-sessions", "--no-legend"]).splitlines() for sid in [sid.split()[0]]): raise RuntimeError("Validation VT is already in use") (DATA / "original-session.json").write_text(json.dumps({"session":active,"vt":vt})+"\n") (DATA / "context.json").unlink(missing_ok=True) config=DATA/"profile/config/ferese";config.mkdir(parents=True,exist_ok=True) (config/"config.kdl").write_text(''' xwayland { enabled #false; } status { keybinding-guide #false; } window-rule app-id="dev.ferese.RimeValidation" floating=#true width=880 height=560 theme { typography { font-family "MiSans"; }; } output-profile "validation" layout="extend" { output "DP-4" mode="3840x2160@60" scale=1.5 { position 0 0; } output "DP-5" mode="3840x2160@160" scale=1.5 { position 2560 0; } output "DP-6" mode="3840x2160@144" scale=1.5 { position 5120 0; } output "HDMI-A-2" required=#false enabled=#false } ''') # Install the return path before anything may acquire the physical seat. subprocess.run(["sudo","-n","systemd-run",f"--unit={RETURN}",f"--on-active={return_seconds}s", "--timer-property=RemainAfterElapse=no","--collect", "/usr/bin/loginctl","activate",active],check=True) env={"XDG_SEAT":"seat0","XDG_VTNR":vt,"XDG_SESSION_TYPE":"wayland","XDG_SESSION_CLASS":"user", "XDG_RUNTIME_DIR":str(RUNTIME),"XDG_CURRENT_DESKTOP":"Ferese","XDG_SESSION_DESKTOP":"ferese", "FERESE_SESSION_IMPORT_ENV":"0","FERESE_ENABLE_INPUT_METHOD":"1","FERESE_ENABLE_SCREENCOPY":"1", "FERESE_RIME_CONTROL":str(RUNTIME/"rime-control.sock"),"LIBSEAT_BACKEND":"logind", "FERESE_ENABLE_TEST_POINTER":"1", "GTK_IM_MODULE":"wayland","QT_IM_MODULE":"wayland","GDK_BACKEND":"wayland", "GSETTINGS_BACKEND":"keyfile","GTK_A11Y":"none","NO_AT_BRIDGE":"1","IBUS_ADDRESS":""} env.update({f"XDG_{kind.upper()}_HOME":str(DATA/"profile"/kind) for kind in ("config","cache","data","state")}) args=["sudo","-n","systemd-run","--collect",f"--unit={MAIN}",f"--uid={USER}","--service-type=exec", "--property=PAMName=login",f"--property=TTYPath=/dev/tty{vt}","--property=StandardInput=tty", "--property=StandardOutput=journal","--property=StandardError=journal","--property=TTYReset=yes", "--property=RuntimeMaxSec=15min","--property=TimeoutStopSec=8", f"--property=ExecStopPost=+/usr/bin/python3 {Path(__file__)} cleanup",f"--working-directory={ROOT}"] args += [f"--setenv={k}={v}" for k,v in env.items()] # pam_systemd sets its standard runtime directory. Restore the private one # before starting the private bus, including its activation environment. subprocess.run(args+["/usr/bin/env",f"XDG_RUNTIME_DIR={RUNTIME}","/usr/bin/dbus-run-session","--","/usr/bin/python3",str(Path(__file__)),"compositor"],check=True) # PAM/logind supplies a distinct session; activate it without terminating GNOME. for _ in range(100): candidates=[line.split()[0] for line in run(["loginctl","list-sessions","--no-legend"]).splitlines()] found=[sid for sid in candidates if run(["loginctl","show-session",sid,"-p","VTNr","--value"])==vt] if found: (DATA/"original-session.json").write_text(json.dumps({"session":active,"vt":vt,"validation_session":found[0]})+"\n") subprocess.run(["sudo","-n","loginctl","activate",found[0]],check=True) print(json.dumps({"validation_session":found[0],"original_session":active,"return_timer_seconds":return_seconds}));return time.sleep(.1) raise RuntimeError("The separate logind session did not appear; return timer remains armed") def clients(): # These clients use the public Wayland socket; shell capabilities never leak. env={k:v for k,v in os.environ.items() if k not in ("WAYLAND_SOCKET","FERESE_SHELL_CONTROL_SOCKET","NOTIFY_SOCKET","WATCHDOG_USEC")} env["WAYLAND_DISPLAY"]=env["FERESE_PUBLIC_WAYLAND_DISPLAY"] sock=socket.socket(socket.AF_UNIX);sock.connect(str(RUNTIME/env["WAYLAND_DISPLAY"])) pid,_,_=struct.unpack("3i",sock.getsockopt(socket.SOL_SOCKET,socket.SO_PEERCRED,12));sock.close() env["FERESE_TEST_DRM_PID"]=str(pid) (DATA/"context.json").write_text(json.dumps({"socket":env["FERESE_SOCKET"],"wayland":env["WAYLAND_DISPLAY"], "bus":env["DBUS_SESSION_BUS_ADDRESS"],"compositor_pid":pid},indent=2)+"\n") def unit(name,command,ime=False): args=["systemd-run","--user","--collect",f"--unit={name}",f"--service-type={'notify' if ime else 'exec'}", f"--working-directory={ROOT}","--property=MemoryMax=1G"] if ime: args += ["--property=Restart=on-failure","--property=RestartSec=1","--property=StartLimitIntervalSec=60", "--property=StartLimitBurst=5","--property=WatchdogSec=10","--property=WatchdogSignal=SIGKILL", "--property=TimeoutStartSec=45","--property=TimeoutStopSec=3"] # Only the explicitly constructed session environment enters the user unit. allowed={k:v for k,v in env.items() if k.startswith(("XDG_","FERESE_","GTK_","QT_","GDK_")) or k in ("WAYLAND_DISPLAY","DBUS_SESSION_BUS_ADDRESS","GSETTINGS_BACKEND","NO_AT_BRIDGE","IBUS_ADDRESS","PATH","HOME","LANG")} subprocess.run(args+[f"--setenv={k}={v}" for k,v in allowed.items()]+command,env=manager_env(),check=True) unit(IME,[str(ROOT/"target/debug/ferese-rime"),"--user-data",str(DATA/"rime-user")],True) unit(APP,["/usr/bin/gjs",str(ROOT/"scripts/tests/fixtures/rime-entry.js"),str(DATA/"entry-state.json")]) def cleanup(): args=["systemctl","--user","stop",APP,IME] if subprocess.run(["systemctl","--user","show",LOCK,"-p","LoadState","--value"], env=manager_env(),text=True,capture_output=True).stdout.strip()=="loaded":args.append(LOCK) if os.getuid()==0: args=["runuser","-u",USER,"--","env",f"XDG_RUNTIME_DIR=/run/user/{UID}", f"DBUS_SESSION_BUS_ADDRESS=unix:path=/run/user/{UID}/bus",*args] subprocess.run(args,env=manager_env(),capture_output=True) def original(): return json.loads((DATA/"original-session.json").read_text())["session"] def real_lock(): c=json.loads((DATA/"context.json").read_text()) env={"XDG_RUNTIME_DIR":str(RUNTIME),"WAYLAND_DISPLAY":c["wayland"],"DBUS_SESSION_BUS_ADDRESS":c["bus"], "FERESE_SOCKET":c["socket"],"XDG_CURRENT_DESKTOP":"Ferese","XDG_SESSION_TYPE":"wayland"} env.update({f"XDG_{kind.upper()}_HOME":str(DATA/"profile"/kind) for kind in ("config","cache","data","state")}) subprocess.run(["systemd-run","--user","--collect",f"--unit={LOCK}","--service-type=exec", "--property=MemoryMax=1G",*[f"--setenv={k}={v}" for k,v in env.items()], "/usr/bin/ferese-lock","--foreground"],env=manager_env(),check=True) def stop(): subprocess.run(["sudo","-n","loginctl","activate",original()],check=True) subprocess.run(["sudo","-n","systemctl","stop",MAIN],check=True) subprocess.run(["sudo","-n","systemctl","stop",f"{RETURN}.timer"],capture_output=True) record=json.loads((DATA/"original-session.json").read_text());sid=record.get("validation_session") if sid and sid!=record["session"]: # PAM may leave its private bus in the abandoned session scope. info=subprocess.run(["loginctl","show-session",sid,"-p","VTNr","--value"],text=True,capture_output=True) if info.returncode==0 and info.stdout.strip()==record["vt"]: subprocess.run(["sudo","-n","loginctl","terminate-session",sid],check=True) subprocess.run(["sudo","-n","loginctl","kill-session","--signal=KILL",sid],capture_output=True) if __name__ == "__main__": parser=argparse.ArgumentParser();parser.add_argument("action",choices=["start","stop","cleanup","clients","compositor","return","status","real-lock"]) action=parser.parse_args().action if action=="compositor": os.environ["XDG_RUNTIME_DIR"]=str(RUNTIME) os.environ.pop("DISPLAY",None);os.environ.pop("WAYLAND_DISPLAY",None) os.execv(str(ROOT/"target/debug/ferese"),["ferese","--backend","drm","--supervise-client","--grant-effects","--grant-shell-control","--","/usr/bin/bash",str(ROOT/"scripts/rime-drm-shell")]) elif action=="return":subprocess.run(["sudo","-n","loginctl","activate",original()],check=True) elif action=="status":print(run(["systemctl","show",MAIN,"-p","ActiveState","-p","MainPID"])) elif action=="real-lock":real_lock() else:globals()[action]()