<!-- 旧版本 -->
<interface name="org.example.Calculator.v1">
  <method name="Add">
    <arg name="a" type="d" direction="in"/>
    <arg name="b" type="d" direction="in"/>
    <arg name="result" type="d" direction="out"/>
  </method>
</interface>

<!-- 新版本（保持向后兼容） -->
<interface name="org.example.Calculator.v2">
  <method name="Add">
    <arg name="a" type="d" direction="in"/>
    <arg name="b" type="d" direction="in"/>
    <arg name="result" type="d" direction="out"/>
  </method>
  <!-- 新增方法 -->
  <method name="AddMultiple">
    <arg name="numbers" type="ad" direction="in"/>
    <arg name="result" type="d" direction="out"/>
  </method>
</interface>

<interface name="org.example.Calculator">
  <property name="Version" type="s" access="read"/>
  <!-- 其他方法和属性 -->
</interface>

<!-- 好的设计：接口按职责拆分 -->
<interface name="org.example.Device">
  <!-- 设备基本信息 -->
  <property name="Name" type="s" access="read"/>
  <property name="Address" type="s" access="read"/>
</interface>

<interface name="org.example.Device.Connection">
  <!-- 连接管理 -->
  <method name="Connect"/>
  <method name="Disconnect"/>
  <signal name="ConnectionStateChanged">
    <arg name="state" type="s"/>
  </signal>
</interface>

<interface name="org.example.Device.Transfer">
  <!-- 数据传输 -->
  <method name="SendData">
    <arg name="data" type="ay" direction="in"/>
  </method>
  <signal name="DataReceived">
    <arg name="data" type="ay"/>
  </signal>
</interface>

<!-- 坏的设计：一个巨大的接口 -->
<interface name="org.example.Device">
  <property name="Name" type="s" access="read"/>
  <property name="Address" type="s" access="read"/>
  <method name="Connect"/>
  <method name="Disconnect"/>
  <method name="SendData"/>
  <signal name="ConnectionStateChanged"/>
  <signal name="DataReceived"/>
  <!-- ... 几十个方法 ... -->
</interface>

<!-- 好：使用 a{sv} 字典传递可选参数 -->
<method name="Connect">
  <arg name="options" type="a{sv}" direction="in"/>
  <arg name="success" type="b" direction="out"/>
</method>

<!-- 坏：大量固定参数 -->
<method name="Connect">
  <arg name="timeout" type="i" direction="in"/>
  <arg name="retry" type="i" direction="in"/>
  <arg name="auto_reconnect" type="b" direction="in"/>
  <arg name="max_retries" type="i" direction="in"/>
  <!-- 参数增加时难以扩展 -->
</method>

<!-- 好：返回 (bs)，包含成功标志和错误消息 -->
<method name="Connect">
  <arg name="options" type="a{sv}" direction="in"/>
  <arg name="success" type="b" direction="out"/>
  <arg name="error_message" type="s" direction="out"/>
</method>

<!-- 好：返回字典，便于扩展 -->
<method name="GetStatus">
  <arg name="status" type="a{sv}" direction="out"/>
</method>

<!-- 坏：直接抛出错误 -->
<method name="Connect">
  <!-- 如果失败，直接返回 Error，不给客户端处理机会 -->
</method>

<interface name="org.example.FileTransfer">
  <!-- 开始操作，返回任务 ID -->
  <method name="StartTransfer">
    <arg name="source" type="s" direction="in"/>
    <arg name="destination" type="s" direction="in"/>
    <arg name="transfer_id" type="u" direction="out"/>
  </method>

  <!-- 查询状态 -->
  <method name="GetTransferStatus">
    <arg name="transfer_id" type="u" direction="in"/>
    <arg name="progress" type="d" direction="out"/>
    <arg name="status" type="s" direction="out"/>
  </method>

  <!-- 取消操作 -->
  <method name="CancelTransfer">
    <arg name="transfer_id" type="u" direction="in"/>
  </method>

  <!-- 进度信号 -->
  <signal name="TransferProgress">
    <arg name="transfer_id" type="u"/>
    <arg name="progress" type="d"/>
  </signal>

  <signal name="TransferCompleted">
    <arg name="transfer_id" type="u"/>
    <arg name="success" type="b"/>
  </signal>
</interface>

<!-- 好：包含足够的上下文信息 -->
<signal name="DeviceConnected">
  <arg name="device_path" type="o"/>      <!-- 对象路径 -->
  <arg name="device_name" type="s"/>      <!-- 人类可读名称 -->
  <arg name="connection_type" type="s"/>  <!-- 连接类型 -->
  <arg name="properties" type="a{sv}"/>   <!-- 额外属性 -->
</signal>

<!-- 坏：信息不足 -->
<signal name="DeviceConnected">
  <arg name="device_id" type="u"/>  <!-- 只有 ID，客户端需要额外查询 -->
</signal>

<!-- 好：属性声明 EmitsChangedSignal -->
<property name="Volume" type="d" access="readwrite">
  <annotation name="org.freedesktop.DBus.Property.EmitsChangedSignal" value="true"/>
</property>
<property name="CachedData" type="a{sv}" access="read">
  <annotation name="org.freedesktop.DBus.Property.EmitsChangedSignal" value="invalidates"/>
</property>
<property name="ConstantValue" type="s" access="read">
  <annotation name="org.freedesktop.DBus.Property.EmitsChangedSignal" value="const"/>
</property>
<property name="RarelyChanged" type="i" access="read">
  <annotation name="org.freedesktop.DBus.Property.EmitsChangedSignal" value="false"/>
</property>

<!-- 坏：不声明注解，客户端不知道变化策略 -->
<property name="Volume" type="d" access="readwrite"/>

# 高频信号的服务端批量发送
class SensorService(dbus.service.Object):
    def __init__(self, bus_name, path):
        super().__init__(bus_name, path)
        self._pending = []
        GLib.timeout_add(100, self._flush)  # 每 100ms 发送一次
    
    @dbus.service.signal(dbus_interface='org.example.Sensor', signature='a{sv}')
    def ReadingsUpdated(self, readings):
        pass
    
    def add_reading(self, sensor_id, value):
        self._pending.append((sensor_id, value, time.time()))
    
    def _flush(self):
        if self._pending:
            readings = {
                'data': self._pending,
                'count': len(self._pending),
            }
            self.ReadingsUpdated(readings)
            self._pending = []
        return True

<!-- 策略文件：最小权限 -->
<busconfig>
  <!-- 只允许特定用户拥有服务名称 -->
  <policy user="my-service">
    <allow own="org.example.MyService"/>
  </policy>

  <!-- 只允许调用必要的接口 -->
  <policy context="default">
    <allow send_destination="org.example.MyService"
           send_interface="org.example.MyService.PublicAPI"/>
    <deny send_destination="org.example.MyService"
          send_interface="org.example.MyService.InternalAPI"/>
  </policy>
</busconfig>

#!/usr/bin/env python3
"""D-Bus 服务的输入验证"""

import dbus
import dbus.service
import re

class MyService(dbus.service.Object):
    
    @dbus.service.method(
        dbus_interface='org.example.MyService',
        in_signature='s',
        out_signature='s'
    )
    def ProcessInput(self, user_input):
        # 验证输入
        if not isinstance(user_input, str):
            raise dbus.exceptions.DBusException(
                'org.example.Error.InvalidInput',
                'Input must be a string'
            )
        
        # 长度限制
        if len(user_input) > 1024:
            raise dbus.exceptions.DBusException(
                'org.example.Error.InvalidInput',
                'Input too long (max 1024 chars)'
            )
        
        # 内容验证（例如：只允许字母数字）
        if not re.match(r'^[a-zA-Z0-9_]+$', user_input):
            raise dbus.exceptions.DBusException(
                'org.example.Error.InvalidInput',
                'Invalid characters in input'
            )
        
        # 安全处理
        return f"Processed: {user_input}"

#!/usr/bin/env python3
"""防止 D-Bus 拒绝服务攻击"""

import time
import dbus
import dbus.service
from collections import defaultdict

class RateLimitedService(dbus.service.Object):
    
    def __init__(self, bus_name, path):
        super().__init__(bus_name, path)
        self._call_counts = defaultdict(list)
        self._max_calls_per_minute = 60
    
    def _check_rate_limit(self, sender):
        """检查调用频率限制"""
        now = time.time()
        # 清理过期记录
        self._call_counts[sender] = [
            t for t in self._call_counts[sender]
            if now - t < 60
        ]
        
        if len(self._call_counts[sender]) >= self._max_calls_per_minute:
            raise dbus.exceptions.DBusException(
                'org.example.Error.RateLimited',
                'Too many requests, try again later'
            )
        
        self._call_counts[sender].append(now)
    
    @dbus.service.method(
        dbus_interface='org.example.MyService',
        in_signature='s',
        out_signature='s',
        sender_keyword='sender'
    )
    def ProcessRequest(self, request, sender=None):
        if sender:
            self._check_rate_limit(sender)
        
        return f"Processed: {request}"

# SELinux D-Bus 策略
# /etc/selinux/targeted/src/policy/my-dbus.te

# 允许我的服务拥有 BusName
allow my_service_t system_dbusd_t:dbus { acquire_svc };
allow my_service_t session_bus_type:dbus { send_msg };

# 允许客户端调用我的服务
allow client_t my_service_t:dbus { send_msg };

# 坏：发送大量小消息
for i in range(1000):
    service.UpdateSingleItem(i, data[i])  # 1000 次调用

# 好：批量发送
service.UpdateItemsBatch(data)  # 1 次调用

# 接口定义
# <method name="UpdateItemsBatch">
#   <arg name="items" type="a{ia{sv}}" direction="in"/>
# </method>

# 坏：每次都读取属性
for i in range(100):
    value = props.Get('org.example.Service', 'SomeProperty')

# 好：使用缓存
value = props.GetAll('org.example.Service')
# 后续从字典中读取

# GDBus Proxy 自动缓存属性
# 只需监听 PropertiesChanged 更新缓存

# 坏：顺序同步调用
result1 = service.Method1()  # 阻塞
result2 = service.Method2()  # 阻塞
result3 = service.Method3()  # 阻塞

# 好：并行异步调用
async def call_all():
    results = await asyncio.gather(
        service.Method1_async(),
        service.Method2_async(),
        service.Method3_async()
    )
    return results

# 监控 D-Bus 消息统计
busctl status

# 监控消息延迟
busctl monitor --match="type='method_call'" &
time busctl call org.example.Service /org/example org.example.Interface SlowMethod

# 使用 systemd-cgtop 监控资源
systemd-cgtop

# 跟踪所有消息
busctl monitor

# 跟踪特定服务
busctl monitor org.example.MyService

# 只跟踪信号
busctl monitor --match="type='signal'"

# 只跟踪方法调用
busctl monitor --match="type='method_call'"

# 只跟踪错误
busctl monitor --match="type='error'"

# GLib D-Bus 调试
G_DBUS_DEBUG=all ./my-app
G_DBUS_DEBUG=message ./my-app
G_DBUS_DEBUG=signal ./my-app

# libdbus 调试
DBUS_VERBOSE=1 ./my-app

# systemd D-Bus 调试
SYSTEMD_LOG_LEVEL=debug busctl ...

# 图形化调试
d-feet &           # GNOME D-Bus 浏览器
d-spy &            # GNOME 45+ 推荐
bustle --session & # 消息时序图

#!/usr/bin/env python3
"""D-Bus 服务的单元测试"""

import unittest
import dbus
import dbus.mainloop.glib
from gi.repository import GLib

dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)

class TestCalculatorService(unittest.TestCase):
    
    @classmethod
    def setUpClass(cls):
        """启动测试服务"""
        cls.bus = dbus.SessionBus()
        # 假设服务已启动
        cls.proxy = cls.bus.get_object(
            'org.example.Calculator',
            '/org/example/Calculator'
        )
        cls.iface = dbus.Interface(cls.proxy, 'org.example.Calculator')
    
    def test_add(self):
        """测试加法"""
        result = self.iface.Add(3.0, 4.0)
        self.assertAlmostEqual(result, 7.0)
    
    def test_multiply(self):
        """测试乘法"""
        result = self.iface.Multiply(5.0, 6.0)
        self.assertAlmostEqual(result, 30.0)
    
    def test_history(self):
        """测试历史记录"""
        self.iface.ClearHistory()
        self.iface.Add(1.0, 2.0)
        history = self.iface.GetHistory()
        self.assertGreater(len(history), 0)
    
    def test_properties(self):
        """测试属性"""
        props = dbus.Interface(self.proxy, 'org.freedesktop.DBus.Properties')
        history = props.Get('org.example.Calculator', 'History')
        self.assertIsInstance(history, dbus.Array)

if __name__ == '__main__':
    unittest.main()

#!/usr/bin/env python3
"""D-Bus 集成测试：测试信号"""

import unittest
import dbus
import dbus.mainloop.glib
from gi.repository import GLib

dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)

class TestSignals(unittest.TestCase):
    
    def test_signal_received(self):
        """测试信号接收"""
        bus = dbus.SessionBus()
        loop = GLib.MainLoop()
        received = []
        
        def on_signal(operation, result):
            received.append((operation, result))
            loop.quit()
        
        bus.add_signal_receiver(
            on_signal,
            signal_name='CalculationPerformed',
            dbus_interface='org.example.Calculator',
        )
        
        # 触发信号
        proxy = bus.get_object('org.example.Calculator', '/org/example/Calculator')
        iface = dbus.Interface(proxy, 'org.example.Calculator')
        iface.Add(1.0, 2.0)
        
        # 等待信号
        loop.run()
        
        self.assertEqual(len(received), 1)
        self.assertEqual(received[0][0], 'add')
        self.assertAlmostEqual(received[0][1], 3.0)

if __name__ == '__main__':
    unittest.main()

### D-Bus 服务部署清单

#### 配置文件
- [ ] D-Bus 策略文件已安装到 `/etc/dbus-1/system.d/`
- [ ] systemd 服务文件已安装到 `/etc/systemd/system/`
- [ ] D-Bus 激活文件已安装到 `/usr/share/dbus-1/system-services/`

#### 权限
- [ ] 服务用户已创建（非 root 运行）
- [ ] 策略文件中只允许必要权限
- [ ] SELinux/AppArmor 策略已配置

#### 激活
- [ ] `Type=dbus` 和 `BusName=` 已配置
- [ ] `WantedBy=multi-user.target` 已设置
- [ ] D-Bus 激活文件中 `SystemdService=` 已设置

#### 测试
- [ ] 手动调用方法测试通过
- [ ] 信号发送/接收测试通过
- [ ] 属性读写测试通过
- [ ] 服务激活测试通过（停止服务后调用能自动启动）
- [ ] 错误处理测试通过

#### 监控
- [ ] 日志输出到 systemd journal
- [ ] 关键信号已记录
- [ ] 性能指标已收集