Implement native inhibition and acknowledged session-ending confirmation
This commit is contained in:
Samuel Onoja committed
1 parent
cc441c049e
commit
29a71e9974
27 files changed
+1915
-142
No files matched your search
@@ -0,0 +1,234 @@
|
||||
"""Exercise real inhibitor FD lifetimes and session handshake on an isolated bus.
|
||||
|
||||
FERESE_TEST_INHIBIT=1 python3 scripts/tests/test_inhibit_isolated.py
|
||||
Uses a nested compositor and mock logind; never suspends or ends the host session.
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import signal
|
||||
import socket
|
||||
import struct
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import threading
|
||||
import time
|
||||
import unittest
|
||||
|
||||
REPO = Path(__file__).resolve().parents[2]
|
||||
NAME = "org.freedesktop.impl.portal.desktop.ferese"
|
||||
PATH = "/org/freedesktop/portal/desktop"
|
||||
INTERFACE = "org.freedesktop.impl.portal.Inhibit"
|
||||
|
||||
|
||||
@unittest.skipUnless(os.environ.get("FERESE_TEST_INHIBIT") == "1", "requires a Wayland host")
|
||||
class InhibitIntegration(unittest.TestCase):
|
||||
def test_private_bus_lifetimes_and_query_end(self):
|
||||
result = subprocess.run(["dbus-run-session", "--", sys.executable, __file__, "--private"], capture_output=True, text=True, timeout=40)
|
||||
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
||||
|
||||
|
||||
class IPC:
|
||||
def __init__(self, path):
|
||||
self.socket = socket.socket(socket.AF_UNIX)
|
||||
self.socket.settimeout(3)
|
||||
self.socket.connect(str(path))
|
||||
|
||||
def call(self, command, args=None):
|
||||
request = json.dumps({"version": 1, "id": 1, "type": "command", "command": command, "args": args or {}}).encode()
|
||||
self.socket.sendall(struct.pack(">I", len(request)) + request)
|
||||
def read(size):
|
||||
result = b""
|
||||
while len(result) < size:
|
||||
chunk = self.socket.recv(size - len(result))
|
||||
if not chunk:
|
||||
raise EOFError("IPC connection closed")
|
||||
result += chunk
|
||||
return result
|
||||
response = json.loads(read(struct.unpack(">I", read(4))[0]))
|
||||
if response.get("error"):
|
||||
raise ValueError(response["error"])
|
||||
return response["result"]
|
||||
|
||||
def close(self):
|
||||
self.socket.close()
|
||||
|
||||
|
||||
def private_test():
|
||||
from gi.repository import Gio, GLib
|
||||
processes = []
|
||||
pipes = []
|
||||
pipe_lock = threading.Lock()
|
||||
ready = threading.Event()
|
||||
login_loop = None
|
||||
login_connection = None
|
||||
manager_xml = """<node><interface name="org.freedesktop.login1.Manager">
|
||||
<method name="Inhibit"><arg type="s" direction="in"/><arg type="s" direction="in"/><arg type="s" direction="in"/><arg type="s" direction="in"/><arg type="h" direction="out"/></method>
|
||||
<method name="ListInhibitors"><arg type="a(ssssuu)" direction="out"/></method>
|
||||
<signal name="PrepareForShutdown"><arg type="b"/></signal>
|
||||
</interface></node>"""
|
||||
|
||||
def logind():
|
||||
nonlocal login_loop, login_connection
|
||||
context = GLib.MainContext.new()
|
||||
context.push_thread_default()
|
||||
login_connection = Gio.DBusConnection.new_for_address_sync(os.environ["DBUS_SESSION_BUS_ADDRESS"], Gio.DBusConnectionFlags.AUTHENTICATION_CLIENT | Gio.DBusConnectionFlags.MESSAGE_BUS_CONNECTION, None, None)
|
||||
login_connection.call_sync("org.freedesktop.DBus", "/org/freedesktop/DBus", "org.freedesktop.DBus", "RequestName", GLib.Variant("(su)", ("org.freedesktop.login1", 0)), None, Gio.DBusCallFlags.NONE, 2000, None)
|
||||
def method(connection, sender, object_path, interface, name, params, invocation):
|
||||
if name == "Inhibit":
|
||||
what, app, reason, mode = params.unpack()
|
||||
reader, writer = os.pipe2(os.O_CLOEXEC | os.O_NONBLOCK)
|
||||
descriptors = Gio.UnixFDList.new()
|
||||
index = descriptors.append(writer)
|
||||
with pipe_lock:
|
||||
pipes.append((reader, what, app, reason, mode))
|
||||
invocation.return_value_with_unix_fd_list(GLib.Variant("(h)", (index,)), descriptors)
|
||||
os.close(writer)
|
||||
else:
|
||||
rows = []
|
||||
with pipe_lock:
|
||||
for reader, what, app, reason, mode in pipes:
|
||||
try:
|
||||
if os.read(reader, 1) == b"":
|
||||
continue
|
||||
except BlockingIOError:
|
||||
pass
|
||||
rows.append((what, app, reason, mode, os.getuid(), os.getpid()))
|
||||
invocation.return_value(GLib.Variant("(a(ssssuu))", (rows,)))
|
||||
login_connection.register_object("/org/freedesktop/login1", Gio.DBusNodeInfo.new_for_xml(manager_xml).interfaces[0], method, None, None)
|
||||
login_loop = GLib.MainLoop.new(context, False)
|
||||
ready.set()
|
||||
login_loop.run()
|
||||
context.pop_thread_default()
|
||||
|
||||
thread = threading.Thread(target=logind, daemon=True)
|
||||
thread.start()
|
||||
assert ready.wait(3), "Mock logind did not start"
|
||||
bus = Gio.bus_get_sync(Gio.BusType.SESSION, None)
|
||||
|
||||
def call(interface, method, params=None, signature=None, path=PATH):
|
||||
return bus.call_sync(NAME, path, interface, method, params, GLib.VariantType.new(signature) if signature else None, Gio.DBusCallFlags.NO_AUTO_START, 5000, None).unpack()
|
||||
|
||||
def wait_for(check, message):
|
||||
deadline = time.monotonic() + 4
|
||||
while time.monotonic() < deadline:
|
||||
while GLib.MainContext.default().pending():
|
||||
GLib.MainContext.default().iteration(False)
|
||||
if check():
|
||||
return
|
||||
time.sleep(0.01)
|
||||
raise AssertionError(message)
|
||||
|
||||
def active_fd(index):
|
||||
with pipe_lock:
|
||||
try:
|
||||
return os.read(pipes[index][0], 1) != b""
|
||||
except BlockingIOError:
|
||||
return True
|
||||
|
||||
with tempfile.TemporaryDirectory(prefix="ferese-inhibit-") as directory:
|
||||
root = Path(directory)
|
||||
runtime = root / "runtime"
|
||||
runtime.mkdir(mode=0o700)
|
||||
config = root / "config/ferese/config.kdl"
|
||||
config.parent.mkdir(parents=True)
|
||||
config.write_text("status { keybinding-guide #false; }\n")
|
||||
display = Path(os.environ["WAYLAND_DISPLAY"])
|
||||
if not display.is_absolute():
|
||||
display = Path(os.environ["XDG_RUNTIME_DIR"]) / display
|
||||
env = dict(os.environ, XDG_RUNTIME_DIR=str(runtime), XDG_CONFIG_HOME=str(root / "config"), WAYLAND_DISPLAY=str(display), DBUS_SYSTEM_BUS_ADDRESS=os.environ["DBUS_SESSION_BUS_ADDRESS"])
|
||||
env.pop("WAYLAND_SOCKET", None)
|
||||
log = (root / "test.log").open("w")
|
||||
try:
|
||||
compositor = subprocess.Popen([str(REPO / "target/debug/ferese"), "--backend=nested"], env=env, stdout=log, stderr=log, start_new_session=True)
|
||||
processes.append(compositor)
|
||||
deadline = time.monotonic() + 10
|
||||
while not (runtime / "ferese/control.sock").exists():
|
||||
assert compositor.poll() is None and time.monotonic() < deadline, (root / "test.log").read_text()
|
||||
time.sleep(0.02)
|
||||
backend = subprocess.Popen([str(REPO / "target/debug/xdg-desktop-portal-ferese")], env=env, stdout=log, stderr=log, start_new_session=True)
|
||||
processes.append(backend)
|
||||
deadline = time.monotonic() + 10
|
||||
while True:
|
||||
try:
|
||||
call("org.freedesktop.DBus.Introspectable", "Introspect", signature="(s)")
|
||||
break
|
||||
except GLib.Error:
|
||||
assert backend.poll() is None and time.monotonic() < deadline, (root / "test.log").read_text()
|
||||
time.sleep(0.02)
|
||||
request = PATH + "/request/test/inhibit"
|
||||
try:
|
||||
call(INTERFACE, "Inhibit", GLib.Variant("(ossua{sv})", (request, "test.media", "", 8, {})))
|
||||
raise AssertionError("Untrusted inhibition was accepted")
|
||||
except GLib.Error as error:
|
||||
assert "AccessDenied" in str(error)
|
||||
bus.call_sync("org.freedesktop.DBus", "/org/freedesktop/DBus", "org.freedesktop.DBus", "RequestName", GLib.Variant("(su)", ("org.freedesktop.portal.Desktop", 0)), None, Gio.DBusCallFlags.NONE, 2000, None)
|
||||
admin = IPC(runtime / "ferese/control.sock")
|
||||
def inhibit(path, flags):
|
||||
call(INTERFACE, "Inhibit", GLib.Variant("(ossua{sv})", (path, "test.media", "", flags, {"reason": GLib.Variant("s", "Playing media")})))
|
||||
inhibit(request, 8)
|
||||
assert active_fd(0)
|
||||
assert admin.call("get-session-state")["inhibitors"][0]["flags"] == 8
|
||||
call("org.freedesktop.impl.portal.Request", "Close", path=request)
|
||||
wait_for(lambda: not active_fd(0) and not admin.call("get-session-state")["inhibitors"], "Close retained inhibition")
|
||||
session = PATH + "/session/test/monitor"
|
||||
changes = []
|
||||
closed = []
|
||||
bus.signal_subscribe(NAME, INTERFACE, "StateChanged", PATH, None, Gio.DBusSignalFlags.NONE, lambda *args: changes.append(args[-1].unpack()[1]))
|
||||
bus.signal_subscribe(NAME, "org.freedesktop.impl.portal.Session", "Closed", session, None, Gio.DBusSignalFlags.NONE, lambda *args: closed.append(True))
|
||||
assert call(INTERFACE, "CreateMonitor", GLib.Variant("(ooss)", (PATH + "/request/test/monitor", session, "test.media", "")), "(u)") == (0,)
|
||||
wait_for(lambda: any(value["session-state"] == 1 for value in changes), "Missing initial Running")
|
||||
assert call("org.freedesktop.DBus.Properties", "Get", GLib.Variant("(ss)", ("org.freedesktop.impl.portal.Session", "version")), "(v)", session) == (1,)
|
||||
query = IPC(runtime / "ferese/control.sock")
|
||||
state = query.call("begin-session-end")
|
||||
assert state["query-ready"] is False
|
||||
wait_for(lambda: any(value["session-state"] == 2 for value in changes), "Missing QueryEnd")
|
||||
call(INTERFACE, "QueryEndResponse", GLib.Variant("(o)", (session,)))
|
||||
assert admin.call("get-session-state")["query-ready"] is True
|
||||
approved = state["inhibitor-revision"]
|
||||
inhibit(request, 1)
|
||||
try:
|
||||
query.call("commit-session-end", {"token": state["query-token"], "inhibitor-revision": approved, "force": True})
|
||||
raise AssertionError("Late inhibitor bypassed confirmation")
|
||||
except ValueError:
|
||||
pass
|
||||
current = admin.call("get-session-state")
|
||||
query.call("validate-session-end", {"token": state["query-token"], "inhibitor-revision": current["inhibitor-revision"], "force": True})
|
||||
assert admin.call("get-session-state")["session-state"] == 2
|
||||
query.call("commit-session-end", {"token": state["query-token"], "inhibitor-revision": current["inhibitor-revision"], "force": True})
|
||||
wait_for(lambda: any(value["session-state"] == 3 for value in changes), "Missing Ending")
|
||||
query.call("cancel-session-end", {"token": state["query-token"]})
|
||||
wait_for(lambda: changes[-1]["session-state"] == 1, "Missing cancelled end state")
|
||||
query.close()
|
||||
call("org.freedesktop.impl.portal.Session", "Close", path=session)
|
||||
wait_for(lambda: len(closed) == 1, "Missing Closed")
|
||||
count = len(changes)
|
||||
login_connection.emit_signal(None, "/org/freedesktop/login1", "org.freedesktop.login1.Manager", "PrepareForShutdown", GLib.Variant("(b)", (True,)))
|
||||
wait_for(lambda: admin.call("get-session-state")["session-state"] == 3, "Missing external shutdown state")
|
||||
time.sleep(0.1)
|
||||
while GLib.MainContext.default().pending():
|
||||
GLib.MainContext.default().iteration(False)
|
||||
assert len(changes) == count, "StateChanged appeared after Closed"
|
||||
compositor.terminate()
|
||||
compositor.wait(timeout=5)
|
||||
wait_for(lambda: not active_fd(1), "Compositor death retained logind inhibitor FD")
|
||||
admin.close()
|
||||
print("Inhibit FD, monitor, ACK, late consent and disconnect checks passed")
|
||||
finally:
|
||||
for process in reversed(processes):
|
||||
if process.poll() is None:
|
||||
os.killpg(process.pid, signal.SIGTERM)
|
||||
process.wait(timeout=5)
|
||||
log.close()
|
||||
login_loop.quit()
|
||||
thread.join(timeout=2)
|
||||
for pipe, *_ in pipes:
|
||||
os.close(pipe)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
if "--private" in sys.argv:
|
||||
private_test()
|
||||
else:
|
||||
unittest.main()
|
||||
@@ -54,7 +54,7 @@ def private_bus_checks():
|
||||
raise AssertionError((root / "backend.log").read_text())
|
||||
time.sleep(0.05)
|
||||
node = ET.fromstring(xml)
|
||||
for name in ("ScreenCast", "Settings", "Screenshot", "Wallpaper", "Background", "Usb", "Lockdown", "GlobalShortcuts"):
|
||||
for name in ("ScreenCast", "Settings", "Screenshot", "Wallpaper", "Background", "Usb", "Lockdown", "GlobalShortcuts", "Inhibit"):
|
||||
interface = node.find(f"interface[@name='org.freedesktop.impl.portal.{name}']")
|
||||
assert interface is not None, name
|
||||
expected = ET.parse(f"/usr/share/dbus-1/interfaces/org.freedesktop.impl.portal.{name}.xml").getroot().find("interface")
|
||||
|
||||
Reference in new issue
Block a user