daemon init

data/meson.build

@@ -43,4 +43,12 @@ configure_file(
   install_dir: get_option('datadir') / 'dbus-1' / 'services'
 )
 
+install_data('ru.ximperlinux.TuneIt.Daemon.policy',
+  install_dir: get_option('datadir') / 'polkit-1' / 'actions'
+)
+
+install_data('ru.ximperlinux.TuneIt.Daemon.conf',
+  install_dir: get_option('sysconfdir') / 'dbus-1' / 'system.d'
+)
+
 subdir('icons')

data/ru.ximperlinux.TuneIt.Daemon.conf

+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE busconfig PUBLIC "-//freedesktop//DTD D-BUS Bus Configuration 1.0//EN"
+"http://www.freedesktop.org/standards/dbus/1.0/busconfig.dtd">
+<busconfig>
+  <type>system</type>
+
+  <!-- Только root может владеть сервисом -->
+  <policy user="root">
+    <allow own="ru.ximperlinux.TuneIt.Daemon"/>
+    <allow send_destination="ru.ximperlinux.TuneIt.Daemon"/>
+    <allow send_interface="ru.ximperlinux.TuneIt.DaemonInterface"/>
+  </policy>
+
+  <!-- Остальные пользователи могут вызывать методы -->
+  <policy context="default">
+    <allow send_destination="ru.ximperlinux.TuneIt.Daemon"/>
+    <allow send_interface="ru.ximperlinux.TuneIt.DaemonInterface"/>
+  </policy>
+</busconfig>

data/ru.ximperlinux.TuneIt.Daemon.policy

+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE policyconfig PUBLIC
+ "-//freedesktop//DTD PolicyKit Policy Configuration 1.0//EN"
+ "http://www.freedesktop.org/standards/PolicyKit/1.0/policyconfig.dtd">
+<policyconfig><vendor>Example</vendor>
+  <vendor_url>https://example.com/example</vendor_url><action id="ru.ximperlinux.TuneIt.Daemon.auth">
+    <description gettext-domain="systemd">Authorization</description>
+    <message gettext-domain="tuneit">You need root rights to read and modify system configs.</message>
+    <defaults>
+        <!--These describe the auth level needed to do this.
+            Auth_admin, the current one, requires admin authentication every time.
+            Auth_admin_keep behaves like sudo, saving the password for a few minutes.Allow_inactive allows it to be accessed from SSH etc. Allow_active allows it to be accessed from the desktop.
+            Allow_any is a combo of both.
+        -->
+      <allow_any>auth_admin_keep</allow_any>
+      <allow_inactive>auth_admin_keep</allow_inactive>
+      <allow_active>auth_admin_keep</allow_active>
+    </defaults>
+  </action>
+</policyconfig>

src/daemon.in

+#!@PYTHON@
+
+# tuneit.in
+#
+# Copyright 2024 Unknown
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+#
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+import os
+import sys
+import signal
+import locale
+import gettext
+
+VERSION = '@VERSION@'
+pkgdatadir = '@pkgdatadir@'
+localedir = '@localedir@'
+
+sys.path.insert(1, pkgdatadir)
+signal.signal(signal.SIGINT, signal.SIG_DFL)
+locale.bindtextdomain('tuneit', localedir)
+locale.textdomain('tuneit')
+gettext.install('tuneit', localedir)
+
+if __name__ == '__main__':
+    import gi
+
+    from gi.repository import Gio
+    resource = Gio.Resource.load(os.path.join(pkgdatadir, 'tuneit.gresource'))
+    resource._register()
+
+    from tuneit import daemon
+    sys.exit(daemon.main())

src/daemon.py

+import ast
+import logging
+
+import dbus
+import dbus.mainloop.glib
+import dbus.service
+from gi.repository import GLib
+
+from .settings.backends import root_backend_factory
+
+# Настройка логирования
+logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
+logger = logging.getLogger(__name__)
+
+class Daemon(dbus.service.Object):
+    def __init__(self, bus, object_path):
+        super().__init__(bus, object_path)
+        self.dbus_info = None
+        self.polkit = None
+
+    def _check_polkit_privilege(self, sender, conn, privilege):
+        """
+        Проверяет привилегии с использованием PolicyKit.
+
+        :param sender: Отправитель D-Bus сообщения.
+        :param conn: Соединение D-Bus.
+        :param privilege: Проверяемая привилегия.
+        :return: True, если авторизация успешна; иначе False.
+        """
+        if self.dbus_info is None:
+            self.dbus_info = dbus.Interface(
+                conn.get_object("org.freedesktop.DBus", "/org/freedesktop/DBus"),
+                "org.freedesktop.DBus"
+            )
+        pid = self.dbus_info.GetConnectionUnixProcessID(sender)
+
+        if self.polkit is None:
+            try:
+                bus_object = dbus.SystemBus().get_object(
+                    "org.freedesktop.PolicyKit1",
+                    "/org/freedesktop/PolicyKit1/Authority"
+                )
+                self.polkit = dbus.Interface(bus_object, "org.freedesktop.PolicyKit1.Authority")
+            except Exception as e:
+                logger.error(f"Failed to connect to PolicyKit: {e}")
+                raise
+
+        retry_limit = 3
+        retry_count = 0
+        while retry_count < retry_limit:
+            try:
+                auth_response = self.polkit.CheckAuthorization(
+                    ("unix-process", {"pid": dbus.UInt32(pid, variant_level=1),
+                                      "start-time": dbus.UInt64(0, variant_level=1)}),
+                    privilege, {"AllowUserInteraction": "true"}, dbus.UInt32(1), "", timeout=600
+                )
+                is_auth, _, _ = auth_response
+                return is_auth
+            except dbus.DBusException as e:
+                if e._dbus_error_name == "org.freedesktop.DBus.Error.ServiceUnknown":
+                    retry_count += 1
+                    logger.warning(f"PolicyKit service unavailable, retrying ({retry_count}/{retry_limit})...")
+                    self.polkit = None
+                else:
+                    logger.error(f"DBusException occurred: {e}")
+                    raise
+
+        logger.error("Failed to authorize: PolicyKit service unavailable after retries.")
+        return False
+
+    @dbus.service.method(
+        dbus_interface="ru.ximperlinux.TuneIt.DaemonInterface",
+        in_signature="ssss",
+        out_signature="s",
+        sender_keyword="sender",
+        connection_keyword="conn"
+    )
+    def GetValue(self, backend_name, backend_params, key, gtype, sender=None, conn=None):
+        if not self._check_polkit_privilege(sender, conn, "ru.ximperlinux.TuneIt.Daemon.auth"):
+            raise dbus.DBusException(
+                "org.freedesktop.DBus.Error.AccessDenied",
+                "Permission denied"
+            )
+        try:
+            backend_params = ast.literal_eval(backend_params)
+            backend = root_backend_factory.get_backend(backend_name, backend_params)
+            if backend:
+                return str(backend.get_value(key, gtype))
+        except Exception as e:
+            return dbus.DBusException(
+                "ru.ximperlinux.TuneIt.Daemon", e
+            )
+        return f"backend_name: {backend_name}, params: {backend_params}, key: {key}"
+
+    @dbus.service.method(
+        dbus_interface="ru.ximperlinux.TuneIt.DaemonInterface",
+        in_signature="sssss",
+        out_signature="s",
+        sender_keyword="sender",
+        connection_keyword="conn"
+    )
+    def SetValue(self, backend_name, backend_params, key, value, gtype, sender=None, conn=None):
+        if not self._check_polkit_privilege(sender, conn, "ru.ximperlinux.TuneIt.Daemon.auth"):
+            raise dbus.DBusException(
+                "org.freedesktop.DBus.Error.AccessDenied",
+                "Permission denied"
+            )
+        try:
+            backend_params = ast.literal_eval(backend_params)
+            backend = root_backend_factory.get_backend(backend_name, backend_params)
+            if backend:
+                backend.set_value(key, value, gtype)
+        except Exception as e:
+            return dbus.DBusException(
+                "ru.ximperlinux.TuneIt.Daemon", e
+            )
+        return f"Failed to set value for backend_name: {backend_name}, key: {key}"
+
+    @dbus.service.method(
+        dbus_interface="ru.ximperlinux.TuneIt.DaemonInterface",
+        in_signature="ssss",
+        out_signature="s",
+        sender_keyword="sender",
+        connection_keyword="conn"
+    )
+    def GetRange(self, backend_name, backend_params, key, gtype, sender=None, conn=None):
+        if not self._check_polkit_privilege(sender, conn, "ru.ximperlinux.TuneIt.Daemon.auth"):
+            raise dbus.DBusException(
+                "org.freedesktop.DBus.Error.AccessDenied",
+                "Permission denied"
+            )
+        try:
+            backend_params = ast.literal_eval(backend_params)
+            backend = root_backend_factory.get_backend(backend_name, backend_params)
+            if backend:
+                return str(backend.get_range(key, gtype))
+        except Exception as e:
+            return dbus.DBusException(
+                "ru.ximperlinux.TuneIt.Daemon", e
+            )
+        return f"Failed to get range for backend_name: {backend_name}, key: {key}"
+
+def main():
+    dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
+
+    try:
+        bus = dbus.SystemBus()
+        name = dbus.service.BusName("ru.ximperlinux.TuneIt.Daemon", bus)
+        daemon = Daemon(bus, "/Daemon")
+        logger.info("Service is running...")
+        mainloop = GLib.MainLoop()
+        mainloop.run()
+    except KeyboardInterrupt:
+        logger.info("Service interrupted by user.")
+    except Exception as e:
+        logger.error(f"Error: {e}")
+
+if __name__ == "__main__":
+    main()

src/meson.build

@@ -35,10 +35,20 @@ configure_file(
   install_mode: 'r-xr-xr-x'
 )
 
+configure_file(
+  input: 'daemon.in',
+  output: 'tuneit-daemon',
+  configuration: conf,
+  install: true,
+  install_dir: get_option('sbindir'),
+  install_mode: 'r-xr-xr-x'
+)
+
 tuneit_sources = [
   '__init__.py',
   'main.py',
   'window.py',
+  'daemon.py',
 ]
 
 install_data(tuneit_sources, install_dir: moduledir)

src/settings/__init__.py

 from gi.repository import Adw, Gtk
 
 from .backends import backend_factory
+from .daemon_client import dclient
+
 from .tools.yml_tools import load_modules, merge_categories_by_name
 from .widgets import WidgetFactory
 
@@ -8,6 +10,7 @@ from .widgets import WidgetFactory
 class Setting:
     def __init__(self, setting_data):
         self.name = setting_data['name']
+        self.root = setting_data.get('root', False)
         self.backend = setting_data.get('backend')
         self.params = setting_data.get('params', {})
         self.type = setting_data['type']
@@ -81,7 +84,12 @@ class Setting:
             backend.set_value(self.key, value, self.gtype)
 
     def _get_backend(self):
-        backend = backend_factory.get_backend(self.backend, self.params)
+        if self.root is True:
+            backend = dclient
+            backend.set_backend_name(self.backend)
+            backend.set_backend_params(self.params)
+        else:
+            backend = backend_factory.get_backend(self.backend, self.params)
 
         if not backend:
             print(f"Бекенд {self.backend} не зарегистрирован.")

src/settings/backends.py

@@ -213,3 +213,19 @@ class BackendFactory:
 
 
 backend_factory = BackendFactory()
+
+class RootBackendFactory:
+    def __init__(self):
+        self.backends = {
+            'file': FileBackend,
+        }
+
+    def get_backend(self, backend_name, params=None):
+        backend_class = self.backends.get(backend_name)
+        if backend_class:
+            # Передаем параметры в конструктор бэкенда, если они есть
+            return backend_class(params) if params else backend_class()
+        return None
+
+
+root_backend_factory = RootBackendFactory()

src/settings/daemon_client.py

+import dbus
+import ast
+
+
+class DaemonClient:
+    def __new__(cls, bus_name="ru.ximperlinux.TuneIt.Daemon", object_path="/Daemon"):
+        """
+        Создает экземпляр клиента только в случае, если сервис доступен.
+
+        :param bus_name: Имя D-Bus сервиса.
+        :param object_path: Путь объекта в D-Bus.
+        :return: Экземпляр DaemonClient или None, если сервис недоступен.
+        """
+        try:
+            bus = dbus.SystemBus()
+            bus.get_object(bus_name, object_path)  # Проверка доступности объекта
+            return super(DaemonClient, cls).__new__(cls)
+        except dbus.DBusException:
+            print(f"Service '{bus_name}' is not running.")
+            return None
+
+    def __init__(self, bus_name="ru.ximperlinux.TuneIt.Daemon", object_path="/Daemon"):
+        """
+        Инициализация клиента для взаимодействия с D-Bus сервисом.
+
+        :param bus_name: Имя D-Bus сервиса.
+        :param object_path: Путь объекта в D-Bus.
+        """
+        self.bus_name = bus_name
+        self.object_path = object_path
+        self.bus = dbus.SystemBus()
+        self.proxy = self.bus.get_object(bus_name, object_path)
+        self.interface = dbus.Interface(
+            self.proxy, dbus_interface="ru.ximperlinux.TuneIt.DaemonInterface"
+        )
+        print("dbus client connected")
+
+        self.backend_name = None
+        self.backend_params = None
+
+    def set_backend_name(self, backend_name):
+        """
+        Устанавливает имя backend.
+
+        :param backend_name: Имя backend.
+        """
+        self.backend_name = backend_name
+
+    def set_backend_params(self, backend_params):
+        """
+        Устанавливает параметры backend.
+
+        :param backend_params: Параметры backend в формате JSON.
+        """
+        self.backend_params = str(backend_params)
+
+    def get_value(self, key, gtype):
+        """
+        Вызывает метод GetValue на D-Bus сервисе.
+
+        :param key: Ключ для получения значения.
+        :param gtype: Тип значения.
+        :return: Полученное значение.
+        """
+        try:
+            return ast.literal_eval(str(self.interface.GetValue(self.backend_name, str(self.backend_params), key, gtype)))
+        except dbus.DBusException as e:
+            print(f"Error in GetValue: {e}")
+            return None
+
+    def set_value(self, key, value, gtype):
+        """
+        Вызывает метод SetValue на D-Bus сервисе.
+
+        :param key: Ключ для установки значения.
+        :param value: Устанавливаемое значение.
+        :param gtype: Тип значения.
+        :return: Результат операции.
+        """
+        try:
+            self.interface.SetValue(self.backend_name, str(self.backend_params), key, str(value), gtype)
+        except dbus.DBusException as e:
+            print(f"Error in SetValue: {e}")
+
+    def get_range(self, key, gtype):
+        """
+        Вызывает метод GetRange на D-Bus сервисе.
+
+        :param key: Ключ для получения диапазона.
+        :param gtype: Тип значения.
+        :return: Диапазон значений.
+        """
+        try:
+            return ast.literal_eval(str(self.interface.GetRange(self.backend_name, str(self.backend_params), key, gtype)))
+        except dbus.DBusException as e:
+            print(f"Error in GetRange: {e}")
+            return None
+
+dclient = DaemonClient()

tuneit.spec

@@ -53,6 +53,12 @@ where each setting is defined in separate files for flexibility and extensibilit
 
 %_iconsdir/hicolor/*/apps/*.svg
 
+
+%_sbindir/tuneit-daemon
+
+%_datadir/polkit-1/actions/*ximper*.policy
+%_sysconfdir/dbus-1/system.d/*ximper*.conf
+
 %changelog
 * Tue Dec 17 2024 Roman Alifanov <ximper@altlinux.org> 0.1.0-alt1
 - initial build