1
0

_dbus.py 4.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105
  1. # systemctl-mqtt - MQTT client triggering & reporting shutdown on systemd-based systems
  2. #
  3. # Copyright (C) 2020 Fabian Peter Hammerle <fabian@hammerle.me>
  4. #
  5. # This program is free software: you can redistribute it and/or modify
  6. # it under the terms of the GNU General Public License as published by
  7. # the Free Software Foundation, either version 3 of the License, or
  8. # any later version.
  9. #
  10. # This program is distributed in the hope that it will be useful,
  11. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  12. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  13. # GNU General Public License for more details.
  14. #
  15. # You should have received a copy of the GNU General Public License
  16. # along with this program. If not, see <https://www.gnu.org/licenses/>.
  17. import datetime
  18. import logging
  19. import dbus
  20. _LOGGER = logging.getLogger(__name__)
  21. def get_login_manager() -> dbus.proxies.Interface:
  22. # https://dbus.freedesktop.org/doc/dbus-python/tutorial.html
  23. bus = dbus.SystemBus()
  24. proxy = bus.get_object(
  25. bus_name="org.freedesktop.login1", object_path="/org/freedesktop/login1"
  26. ) # type: dbus.proxies.ProxyObject
  27. # https://freedesktop.org/wiki/Software/systemd/logind/
  28. return dbus.Interface(object=proxy, dbus_interface="org.freedesktop.login1.Manager")
  29. def _log_shutdown_inhibitors(login_manager: dbus.proxies.Interface) -> None:
  30. if _LOGGER.getEffectiveLevel() > logging.DEBUG:
  31. return
  32. found_inhibitor = False
  33. try:
  34. # https://www.freedesktop.org/wiki/Software/systemd/inhibit/
  35. for what, who, why, mode, uid, pid in login_manager.ListInhibitors():
  36. if "shutdown" in what:
  37. found_inhibitor = True
  38. _LOGGER.debug(
  39. "detected shutdown inhibitor %s (pid=%u, uid=%u, mode=%s): %s",
  40. who,
  41. pid,
  42. uid,
  43. mode,
  44. why,
  45. )
  46. except dbus.DBusException as exc:
  47. _LOGGER.warning(
  48. "failed to fetch shutdown inhibitors: %s", exc.get_dbus_message()
  49. )
  50. return
  51. if not found_inhibitor:
  52. _LOGGER.debug("no shutdown inhibitor locks found")
  53. def schedule_shutdown(*, action: str, delay: datetime.timedelta) -> None:
  54. # https://github.com/systemd/systemd/blob/v237/src/systemctl/systemctl.c#L8553
  55. assert action in ["poweroff", "reboot"], action
  56. shutdown_datetime = datetime.datetime.now() + delay
  57. # datetime.datetime.isoformat(timespec=) not available in python3.5
  58. # https://github.com/python/cpython/blob/v3.5.9/Lib/datetime.py#L1552
  59. _LOGGER.info(
  60. "scheduling %s for %s", action, shutdown_datetime.strftime("%Y-%m-%d %H:%M:%S")
  61. )
  62. # https://dbus.freedesktop.org/doc/dbus-python/tutorial.html?highlight=signature#basic-types
  63. shutdown_epoch_usec = dbus.UInt64(shutdown_datetime.timestamp() * 10**6)
  64. login_manager = get_login_manager()
  65. try:
  66. # $ gdbus introspect --system --dest org.freedesktop.login1 \
  67. # --object-path /org/freedesktop/login1 | grep -A 1 ScheduleShutdown
  68. # ScheduleShutdown(in s arg_0,
  69. # in t arg_1);
  70. # $ gdbus call --system --dest org.freedesktop.login1 \
  71. # --object-path /org/freedesktop/login1 \
  72. # --method org.freedesktop.login1.Manager.ScheduleShutdown \
  73. # poweroff "$(date --date=10min +%s)000000"
  74. # $ dbus-send --type=method_call --print-reply --system --dest=org.freedesktop.login1 \
  75. # /org/freedesktop/login1 \
  76. # org.freedesktop.login1.Manager.ScheduleShutdown \
  77. # string:poweroff "uint64:$(date --date=10min +%s)000000"
  78. login_manager.ScheduleShutdown(action, shutdown_epoch_usec)
  79. except dbus.DBusException as exc:
  80. exc_msg = exc.get_dbus_message()
  81. if "authentication required" in exc_msg.lower():
  82. _LOGGER.error(
  83. "failed to schedule %s: unauthorized; missing polkit authorization rules?",
  84. action,
  85. )
  86. else:
  87. _LOGGER.error("failed to schedule %s: %s", action, exc_msg)
  88. _log_shutdown_inhibitors(login_manager)
  89. def lock_all_sessions() -> None:
  90. """
  91. $ loginctl lock-sessions
  92. """
  93. _LOGGER.info("instruct all sessions to activate screen locks")
  94. get_login_manager().LockSessions()