fix fatal `org.freedesktop.DBus.Error.InteractiveAuthorizationRequired` when attempting to lock sessions, suggest udev rule

https://github.com/fphammerle/systemctl-mqtt/commit/452940d2516374f1779a98021b10d8cbb83448ef

CHANGELOG.md

@@ -21,7 +21,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   - entity `sensor.[hostname]_unit_system_[unit_name]_active_state`
     for each command-line parameter `--monitor-system-unit [unit_name]`
 - command-line option `--log-level {debug,info,warning,error,critical}`
-- suggest polkit rule when poweroff fails due to
+- suggest polkit rule when poweroff & locking fails due to
   `org.freedesktop.DBus.Error.InteractiveAuthorizationRequired`
   (https://github.com/fphammerle/systemctl-mqtt/issues/67)
 - declare compatibility with `python3.11`, `python3.12` & `python3.13`
@@ -56,6 +56,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Fixed
 - apparmor profile for architectures other than x86_64/amd64
   (`ImportError: Error loading [...]/_gi.cpython-38-aarch64-linux-gnu.so: Permission denied`)
+- fatal `org.freedesktop.DBus.Error.InteractiveAuthorizationRequired` when
+  attempting to lock sessions
 - container image / dockerfile:
   - split `pipenv install` into two stages to speed up image builds
   - `chmod` files copied from host to no longer require `o=rX` perms on host

systemctl_mqtt/_dbus/login_manager.py

@@ -192,10 +192,7 @@ def schedule_shutdown(*, action: str, delay: datetime.timedelta) -> None:
         #       string:poweroff "uint64:$(date --date=10min +%s)000000"
         login_manager.ScheduleShutdown(action=action, time=time)
     except jeepney.wrappers.DBusErrorResponse as exc:
-        if (
-            exc.name == "org.freedesktop.DBus.Error.InteractiveAuthorizationRequired"
-            and exc.data == ("Interactive authentication required.",)
-        ):
+        if exc.name == "org.freedesktop.DBus.Error.InteractiveAuthorizationRequired":
             _log_interactive_authorization_required(
                 action_label="schedule " + action,
                 action_id="org.freedesktop.login1."
@@ -216,4 +213,14 @@ def lock_all_sessions() -> None:
     $ loginctl lock-sessions
     """
     _LOGGER.info("instruct all sessions to activate screen locks")
-    get_login_manager_proxy().LockSessions()
+    login_manager = get_login_manager_proxy()
+    try:
+        login_manager.LockSessions()
+    except jeepney.wrappers.DBusErrorResponse as exc:
+        if exc.name == "org.freedesktop.DBus.Error.InteractiveAuthorizationRequired":
+            _log_interactive_authorization_required(
+                action_label="lock all sessions",
+                action_id="org.freedesktop.login1.lock-sessions",
+            )
+        else:
+            _LOGGER.error("failed to lock all sessions: %s", exc)

tests/test_dbus.py

@@ -254,6 +254,52 @@ def test_lock_all_sessions(caplog):
     assert caplog.records[0].message == "instruct all sessions to activate screen locks"
 
 
+@pytest.mark.parametrize(
+    ("error_name", "error_message", "log_message"),
+    [
+        (
+            "test error",
+            "test message",
+            "[test error] ('test message',)",
+        ),
+        (
+            "org.freedesktop.DBus.Error.InteractiveAuthorizationRequired",
+            "Interactive authentication required.",
+            """interactive authorization required
+
+create /etc/polkit-1/rules.d/50-systemctl-mqtt.rules and insert the following rule:
+polkit.addRule(function(action, subject) {
+    if(action.id === "org.freedesktop.login1.lock-sessions" && subject.user === "{{username}}") {
+        return polkit.Result.YES;
+    }
+});
+""".replace(
+                "{{username}}", getpass.getuser()
+            ),
+        ),
+    ],
+)
+def test_lock_all_sessions_fail(
+    caplog: pytest.LogCaptureFixture,
+    error_name: str,
+    error_message: str,
+    log_message: str,
+) -> None:
+    login_manager_mock = unittest.mock.MagicMock()
+    login_manager_mock.LockSessions.side_effect = DBusErrorResponseMock(
+        name=error_name, data=(error_message,)
+    )
+    with unittest.mock.patch(
+        "systemctl_mqtt._dbus.login_manager.get_login_manager_proxy",
+        return_value=login_manager_mock,
+    ), caplog.at_level(logging.ERROR):
+        systemctl_mqtt._dbus.login_manager.lock_all_sessions()
+    login_manager_mock.LockSessions.assert_called_once()
+    assert len(caplog.records) == 1
+    assert caplog.records[0].levelno == logging.ERROR
+    assert caplog.records[0].message == f"failed to lock all sessions: {log_message}"
+
+
 async def _get_unit_path_mock(  # pylint: disable=unused-argument
     *, service_manager: jeepney.io.asyncio.Proxy, unit_name: str
 ) -> str: