181 lines
11 KiB
Python
181 lines
11 KiB
Python
#!/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]()
|