Add configurable session-scoped global shortcut portal

scripts/tests/test_portal_contracts.py

@@ -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"):
+            for name in ("ScreenCast", "Settings", "Screenshot", "Wallpaper", "Background", "Usb", "Lockdown", "GlobalShortcuts"):
                 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")
@@ -90,6 +90,14 @@ def private_bus_checks():
                 raise AssertionError("Untrusted backend caller succeeded")
             except GLib.Error:
                 pass
+            shortcut_session = PATH + "/session/test/shortcuts"
+            shortcut_handle = PATH + "/request/test/shortcuts"
+            create_shortcuts = GLib.Variant("(oosa{sv})", (shortcut_handle, shortcut_session, "org.test.Shortcuts", {}))
+            try:
+                call("org.freedesktop.impl.portal.GlobalShortcuts", "CreateSession", create_shortcuts)
+                raise AssertionError("Untrusted shortcuts creator succeeded")
+            except GLib.Error as error:
+                assert "AccessDenied" in str(error), error
             original_config = config.read_text()
             try:
                 call("org.freedesktop.DBus.Properties", "Set", GLib.Variant("(ssv)", ("org.freedesktop.impl.portal.Lockdown", "disable-printing", GLib.Variant("b", True))))
@@ -129,6 +137,24 @@ def private_bus_checks():
             config.write_text(valid_policy)
             bus.signal_unsubscribe(subscription)
 
+            response, values = call("org.freedesktop.impl.portal.GlobalShortcuts", "CreateSession", create_shortcuts, "(ua{sv})")
+            assert response == 0 and values == {}
+            response, values = call("org.freedesktop.impl.portal.GlobalShortcuts", "ListShortcuts", GLib.Variant("(oo)", (shortcut_handle, shortcut_session)), "(ua{sv})")
+            assert response == 0 and values == {"shortcuts": []}, values
+            duplicate = [("record", {"description": GLib.Variant("s", "Record screen")})] * 2
+            try:
+                call("org.freedesktop.impl.portal.GlobalShortcuts", "BindShortcuts", GLib.Variant("(ooa(sa{sv})sa{sv})", (shortcut_handle, shortcut_session, duplicate, "", {})))
+                raise AssertionError("Duplicate shortcut IDs accepted")
+            except GLib.Error as error:
+                assert "InvalidArgs" in str(error), error
+            bus.call_sync(NAME, shortcut_session, "org.freedesktop.impl.portal.Session", "Close", None, None,
+                          Gio.DBusCallFlags.NO_AUTO_START, 5000, None)
+            try:
+                call("org.freedesktop.impl.portal.GlobalShortcuts", "ListShortcuts", GLib.Variant("(oo)", (shortcut_handle, shortcut_session)))
+                raise AssertionError("Closed shortcut session remains accessible")
+            except GLib.Error as error:
+                assert "AccessDenied" in str(error), error
+
             usb_params = GLib.Variant("(oss a(sa{sv}a{sv}) a{sv})".replace(" ", ""),
                 (PATH + "/request/test/usb", "", "org.test.App", [], {}))
             response, values = call("org.freedesktop.impl.portal.Usb", "AcquireDevices", usb_params, "(ua{sv})")

scripts/tests/test_shortcuts_isolated.py

@@ -0,0 +1,95 @@
+"""Exercise shortcut connection lifetimes in an isolated nested compositor."""
+import json
+import os
+from pathlib import Path
+import signal
+import socket
+import struct
+import subprocess
+import tempfile
+import time
+import unittest
+
+REPO = Path(__file__).resolve().parents[2]
+
+
+@unittest.skipUnless(os.environ.get("FERESE_TEST_SHORTCUTS") == "1", "requires a Wayland host and a built compositor")
+class ShortcutConnections(unittest.TestCase):
+    def test_disconnect_conflicts_and_configuration_reload(self):
+        host_runtime = Path(os.environ["XDG_RUNTIME_DIR"])
+        host_display = Path(os.environ["WAYLAND_DISPLAY"])
+        if not host_display.is_absolute():
+            host_display = host_runtime / host_display
+        with tempfile.TemporaryDirectory(prefix="ferese-shortcut-test-") 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("")
+            log = (root / "compositor.log").open("w")
+            env = dict(os.environ, XDG_RUNTIME_DIR=str(runtime), XDG_CONFIG_HOME=str(root / "config"), WAYLAND_DISPLAY=str(host_display))
+            env.pop("WAYLAND_SOCKET", None)
+            env.pop("FERESE_SHELL_CONTROL_SOCKET", None)
+            process = subprocess.Popen([str(REPO / "target/debug/ferese"), "--backend=nested"], env=env,
+                                       stdout=log, stderr=log, start_new_session=True)
+            connections = []
+            try:
+                deadline = time.monotonic() + 15
+                while not (runtime / "ferese/control.sock").exists():
+                    if process.poll() is not None or time.monotonic() >= deadline:
+                        self.fail((root / "compositor.log").read_text())
+                    time.sleep(0.02)
+
+                def connect():
+                    connection = socket.socket(socket.AF_UNIX)
+                    connection.settimeout(2)
+                    connection.connect(str(runtime / "ferese/control.sock"))
+                    connections.append(connection)
+                    return connection
+
+                def call(connection, command, args=None):
+                    data = json.dumps({"version": 1, "id": 1, "type": "command", "command": command, "args": args}).encode()
+                    connection.sendall(struct.pack(">I", len(data)) + data)
+
+                    def receive(length):
+                        data = b""
+                        while len(data) < length:
+                            chunk = connection.recv(length - len(data))
+                            if not chunk:
+                                raise AssertionError("IPC connection closed")
+                            data += chunk
+                        return data
+
+                    length, = struct.unpack(">I", receive(4))
+                    self.assertLessEqual(length, 1024 * 1024)
+                    return json.loads(receive(length))
+
+                first = connect()
+                second = connect()
+                shortcut = [{"id": "record", "trigger": "Ctrl+Alt+a"}]
+                self.assertNotIn("error", call(first, "portal-shortcuts-register", shortcut))
+                conflict = call(second, "portal-shortcuts-register", shortcut)
+                self.assertEqual(conflict["error"]["code"], "shortcut_conflict")
+                self.assertTrue(call(second, "portal-shortcuts-poll")["result"]["closed"])
+                first.close()
+                deadline = time.monotonic() + 3
+                while "error" in call(second, "portal-shortcuts-register", shortcut):
+                    self.assertLess(time.monotonic(), deadline, "disconnected owner kept its binding")
+                    time.sleep(0.02)
+                self.assertEqual(call(second, "portal-shortcuts-poll")["result"]["shortcuts"], shortcut)
+                config.write_text('binding keys="Ctrl+Alt+a" action="none"\n')
+                deadline = time.monotonic() + 5
+                while call(second, "portal-shortcuts-poll")["result"]["shortcuts"]:
+                    self.assertLess(time.monotonic(), deadline, "configuration did not revoke conflicting shortcut")
+                    time.sleep(0.05)
+            finally:
+                for connection in connections:
+                    connection.close()
+                os.killpg(process.pid, signal.SIGTERM)
+                process.wait(timeout=5)
+                log.close()
+
+
+if __name__ == "__main__":
+    unittest.main()