正确的 Python DBus 连接语法？

Question

我在连接 dbus 时遇到问题：

    try:
        logging.debug("Attempting to connect to D-Bus.")
        self.bus = SessionBus()
        self.keepass_service = self.bus.get("org.keepassxc.KeePassXC.MainWindow", "/org/keepassxc/KeePassXC/MainWindow")
        # self.keepass_service = self.bus.get("org.keepassxc.KeePassXC", "/org/keepassxc/KeePassXC/")
        # self.keepass_service = self.bus.get("org.keepassxc.KeePassXC.MainWindow")

Dbus.Listnames 显示：

    $ dbus-send --print-reply --dest=org.freedesktop.DBus --type=method_call /org/freedesktop/DBus org.freedesktop.DBus.ListNames
method return time=1729375987.604568 sender=org.freedesktop.DBus -> destination=:1.826 serial=3 reply_serial=2
   array [
      string "org.freedesktop.DBus"
      string ":1.469"
      string "org.freedesktop.Notifications"
      string "org.freedesktop.PowerManagement"
      string ":1.7"
      string "org.keepassxc.KeePassXC.MainWindow"

此版本产生此错误：

    self.keepass_service = self.bus.get("org.keepassxc.KeePassXC.MainWindow", "/org/keepassxc/KeePassXC/MainWindow")


ERROR:root:Error message: g-dbus-error-quark: GDBus.Error:org.freedesktop.DBus.Error.UnknownObject: No such object path '/org/keepassxc/KeePassXC/MainWindow' (41)

此版本产生此错误：

    self.keepass_service = self.bus.get("org.keepassxc.KeePassXC", "/org/keepassxc/KeePassXC/")


(process:607644): GLib-GIO-CRITICAL **: 16:18:39.599: g_dbus_connection_call_sync_internal: assertion 'object_path != NULL && g_variant_is_object_path (object_path)' failed
ERROR:root:Failed to connect to KeePassXC D-Bus interface.
ERROR:root:Error message: 'no such object; you might need to pass object path as the 2nd argument for get()'

我尝试添加时间延迟，以防出现竞争条件。 我尝试过使用已经运行的 keepassxc 实例。 我不知道下一步该去哪里？

这是完整上下文的代码：

from pydbus import SessionBus
import logging
import os
import subprocess
from gi.repository import GLib

import time




# Set up logging configuration
logging.basicConfig(level=logging.DEBUG)  # Set logging level to debug

class KeePassXCManager:
    def __init__(self, db_path, password=None, keyfile=None, appimage_path=None):
        logging.debug("Initializing KeePassXCManager")

        self.db_path = db_path
        self.password = password
        self.keyfile = keyfile
        self.kp = None
        self.keepass_command = []

        # Set default path to the KeePassXC AppImage in ~/Applications
        self.appimage_path = appimage_path or os.path.expanduser("~/Applications/KeePassXC.appimage")
        logging.debug(f"AppImage path set to: {self.appimage_path}")

        # Determine the KeePassXC launch command
        self._set_keepassxc_command()  
        self._ensure_keepassxc_running()

        # Set up the D-Bus connection to KeePassXC
        self.bus = SessionBus()
        self.keepass_service = None
        self._connect_to_dbus()

        # Open the database once the manager is initialized
        if not self.open_database():
            logging.error("Failed to open the database during initialization.")
      
    def _set_keepassxc_command(self):
        """Sets the command to launch KeePassXC."""
        try:
            if self._is_keepassxc_installed():
                logging.info("Using installed KeePassXC version.")
                self.keepass_command = ["keepassxc"]
            elif os.path.isfile(self.appimage_path) and os.access(self.appimage_path, os.X_OK):
                logging.info(f"KeePassXC AppImage is executable at {self.appimage_path}")
                self.keepass_command = [self.appimage_path]
            else:
                logging.error("KeePassXC is not installed or AppImage is not executable.")
                raise RuntimeError("KeePassXC is not installed. Please install it or provide a valid AppImage.")

            logging.debug(f"Final KeePassXC command set: {self.keepass_command}")
        except Exception as e:
            logging.error(f"Error setting KeePassXC command: {e}")
            raise

    def _is_keepassxc_installed(self):
        """Checks if KeePassXC is installed on the system."""
        logging.debug("Checking if KeePassXC is installed via package manager")
        try:
            result = subprocess.run(["which", "keepassxc"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
            if result.returncode == 0:
                logging.info(f"KeePassXC found at {result.stdout.decode().strip()}")
                return True
            else:
                logging.warning("KeePassXC is not installed via package manager.")
                return False
        except Exception as e:
            logging.error(f"Error checking KeePassXC installation: {e}")
            return False

    def _ensure_keepassxc_running(self):
        """Checks if KeePassXC is running and starts it if not."""
        logging.debug("Checking if KeePassXC is running")
        try:
            # Check if KeePassXC is running using pgrep
            result = subprocess.run(["pgrep", "-x", "keepassxc"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)

            if result.returncode != 0:
                logging.info("KeePassXC is not running. Starting KeePassXC.")
                # Start KeePassXC
                subprocess.Popen(self.keepass_command)
                # Optionally, wait for a short time to allow KeePassXC to start
                GLib.idle_add(lambda: None)  # Allows the GUI to initialize
            else:
                logging.info("KeePassXC is already running.")

        except Exception as e:
            logging.error(f"Error checking or starting KeePassXC: {e}")

    def _construct_open_command(self):
        """Constructs the command to open the KeePassXC database."""
        command = [self.keepass_command[0], self.db_path]
        
        if self.password:
            command.append("--pw-stdin")
            logging.debug(f"Command includes password for opening database: {self.db_path}")

        if self.keyfile:
            command.append(f"--keyfile={self.keyfile}")
            logging.debug(f"Command includes keyfile for opening database: {self.keyfile}")

        logging.debug(f"Final command to open KeePassXC database: {command}")

        return command if self.password or self.keyfile else None  

    def _clear_sensitive_data(self):
        """Clears sensitive data from memory."""
        logging.debug("Clearing sensitive data from memory")
        self.password = None
        self.keyfile = None
        self.db_path = None
    
    def _connect_to_dbus(self):
        """Connects to the KeePassXC D-Bus interface."""
        try:
            logging.debug("Attempting to connect to D-Bus.")
            self.bus = SessionBus()
            # self.keepass_service = self.bus.get("org.keepassxc.KeePassXC.MainWindow", "/org/keepassxc/KeePassXC/MainWindow")
            self.keepass_service = self.bus.get("org.keepassxc.KeePassXC", "/org/keepassxc/KeePassXC/")
            # self.keepass_service = self.bus.get("org.keepassxc.KeePassXC.MainWindow")
            # self.keepass_service = self.bus.get("org.KeePassXC.MainWindow", "/org/KeePassXC/MainWindow")


            if self.keepass_service:
                logging.info("Successfully connected to KeePassXC D-Bus interface.")
            else:
                logging.error("KeePassXC D-Bus interface is not available.")
            
        except Exception as e:
            logging.error("Failed to connect to KeePassXC D-Bus interface.")
            logging.error(f"Error message: {e}")
            services = self.bus.get_services()
            logging.error(f"Available D-Bus services: {services}")


    def open_database(self):
        """Opens the KeePassXC database using D-Bus."""
        try:
            if not self.keepass_service:
                logging.error("KeePassXC D-Bus service is not available.")
                return False

            logging.info(f"Opening database: {self.db_path}")

            # Prepare parameters for the D-Bus call
            password = self.password or ""
            keyfile = self.keyfile or ""

            # Call the D-Bus method with parameters directly
            response = self.keepass_service.openDatabase(self.db_path, password, keyfile)
            
            if response:
                logging.info("Database opened successfully via D-Bus.")
                return True
            else:
                logging.error("Failed to open database via D-Bus.")
                return False

        except Exception as e:
            logging.error(f"An error occurred while opening the database: {e}")
            return False

    def unlock_database(self):
        """Unlocks the KeePassXC database with the password via D-Bus."""
        try:
            if not self.keepass_service:
                logging.error("KeePassXC D-Bus service is not available.")
                return False

            logging.info("Unlocking database with the provided password.")
            response = self.keepass_service.unlockDatabase(self.password)

            if response:
                logging.info("Database unlocked successfully via D-Bus.")
                return True
            else:
                logging.error("Failed to unlock database via D-Bus.")
                return False
        except Exception as e:
            logging.error(f"An error occurred while unlocking the database: {e}")
            return False

Answer 1

您假设对象路径始终遵循服务本身的命名。情况并非总是如此 - 服务可以导出许多不同的对象路径，并且并不严格需要遵循任何命名样式（即，没有强制规则要求所有对象路径以服务名称开头，更不用说有是一个完全匹配的；两者都只是约定）。

查看busctl或D-Spy（或旧的D-Feet），KeePassXC似乎不遵循任何通常的D-Bus对象命名约定，并且它公开的唯一对象位于路径

/keepassxc

。 

（KeePassXC 是一个基于 Qt 的应用程序，其中许多众所周知不遵循使用服务名称作为对象路径的“基础”的一般 D-Bus 约定；相反，旧的 KDE3 DCOP（D-Bus 之前的版本） ）所有对象直接以

/

 为根的样式在 Qt 程序中仍然很常见。）

$ busctl --user tree org.keepassxc.KeePassXC.MainWindow
└─ /keepassxc

$ gdbus introspect -e -d org.keepassxc.KeePassXC.MainWindow -o / -r -p
node / {
  node /keepassxc {
  };
};