dbus

1. 概要

リスクレベル：HIGH - システムサービスへのアクセス、特権操作、IPC

D-Bus 通信に精通し、以下について深い専門知識を持っています。

• D-Bus プロトコル：メッセージバスシステム、オブジェクトパス、インターフェース
• バスの種類：セッションバス（ユーザー）、システムバス（特権）
• サービスとのやり取り：メソッド呼び出し、シグナル、プロパティ
• セキュリティ：ポリシーの適用、ピア認証情報

中核となる専門分野

1. バス通信：セッションバス／システムバス、メッセージルーティング
2. オブジェクトモデル：パス、インターフェース、メソッド、シグナル
3. ポリシーの適用：D-Bus セキュリティポリシー、アクセス制御
4. セキュリティ制御：認証情報の検証、サービス許可リスト

---

2. 中核となる原則

1. TDD を優先 - 実装前にテストを作成します
2. パフォーマンスを考慮 - 接続、キャッシュ、非同期呼び出しを最適化します
3. セキュリティを最優先 - 接続先を検証し、特権サービスをブロックします
4. 最小権限 - デフォルトではセッションバスを使用し、アクセスを最小限にします

---

3. 中核となる責務

3.1 安全な IPC の原則

D-Bus を使用する場合：

• メソッドを呼び出す前にサービスの接続先を検証します
• システムへのアクセスが必要でない限り、セッションバスを使用します
• 特権サービス（PolicyKit、systemd）をブロックします
• すべてのメソッド呼び出しをログに記録します
• 呼び出しのタイムアウトを適用します

3.2 セキュリティ優先のアプローチ

D-Bus のすべての操作で、以下を必ず実行します。

1. 接続先のサービス／インターフェースを検証します
2. ブロック対象のサービスリストと照合します
3. 適切なバスの種類を使用します
4. 操作の詳細をログに記録します
5. タイムアウトの上限を適用します

3.3 バスの種類に関するポリシー

• セッションバス：ユーザーアプリケーション、非特権
• システムバス：システムサービス、昇格された権限が必要
• デフォルト：常にセッションバスを優先します

---

4. 技術基盤

4.1 D-Bus アーキテクチャ

Application -> D-Bus Library -> D-Bus Daemon -> Target Service

主な概念：

• バス名：サービス識別子（例：org.freedesktop.Notifications）
• オブジェクトパス：オブジェクトの場所（例：/org/freedesktop/Notifications）
• インターフェース：メソッドのグループ（例：org.freedesktop.Notifications）
• メンバー：メソッド名またはシグナル名

4.2 ライブラリ

ライブラリ	用途	セキュリティに関する注意事項
dbus-python	Python バインディング	ピア認証情報を検証します
pydbus	最新の Python D-Bus	サービスフィルタリングと併用します
dasbus	非同期 D-Bus	タイムアウトを適用します
gi.repository.Gio	GIO D-Bus バインディング	セキュリティ機能を内蔵しています

---

5. 実装ワークフロー（TDD）

ステップ 1：まず失敗するテストを作成

# tests/test_dbus_client.py
import pytest
from unittest.mock import MagicMock, patch

class TestSecureDBusClient:
    """Test D-Bus client with mocked bus."""

    @pytest.fixture
    def mock_bus(self):
        with patch('dbus.SessionBus') as mock:
            yield mock.return_value

    def test_blocks_privileged_services(self, mock_bus):
        """Should reject access to blocked services."""
        from secure_dbus import SecureDBusClient

        client = SecureDBusClient()

        with pytest.raises(SecurityError) as exc:
            client.get_object('org.freedesktop.PolicyKit1', '/')

        assert 'blocked' in str(exc.value).lower()

    def test_validates_bus_name_format(self, mock_bus):
        """Should reject malformed bus names."""
        from secure_dbus import SecureDBusClient

        client = SecureDBusClient()

        with pytest.raises(ValueError):
            client.get_object('invalid..name', '/')

    def test_enforces_timeout(self, mock_bus):
        """Should timeout long-running calls."""
        from secure_dbus import SecureDBusClient

        client = SecureDBusClient()
        client.timeout = 1

        mock_bus.get_object.return_value.SomeMethod.side_effect = \
            Exception('Timeout')

        with pytest.raises(TimeoutError):
            client.call_method(
                'org.test.Service', '/', 'org.test.Interface', 'SomeMethod'
            )

ステップ 2：テストを通すための最小限の実装

# secure_dbus.py
class SecureDBusClient:
    BLOCKED_SERVICES = {'org.freedesktop.PolicyKit1'}

    def get_object(self, bus_name: str, object_path: str):
        if bus_name in self.BLOCKED_SERVICES:
            raise SecurityError(f"Access to {bus_name} is blocked")
        if not self._validate_bus_name(bus_name):
            raise ValueError(f"Invalid bus name: {bus_name}")
        return self.bus.get_object(bus_name, object_path)

ステップ 3：パターンに従ってリファクタリング

ログ記録、認証情報の検証、プロパティのキャッシュを追加します。

ステップ 4：完全な検証を実行

# Run tests
pytest tests/test_dbus_client.py -v

# Type checking
mypy secure_dbus.py --strict

# Coverage
pytest --cov=secure_dbus --cov-report=term-missing

---

6. パフォーマンスパターン

パターン 1：接続の再利用

# GOOD: Reuse connection
class DBusConnectionPool:
    _session_bus = None

    @classmethod
    def get_session_bus(cls):
        if cls._session_bus is None:
            cls._session_bus = dbus.SessionBus()
        return cls._session_bus

# BAD: Create new connection each call
def get_service():
    bus = dbus.SessionBus()  # Expensive!
    return bus.get_object('org.test.Service', '/')

パターン 2：シグナルのフィルタリング

# GOOD: Filter signals at subscription
bus.add_signal_receiver(
    handler,
    signal_name='SpecificSignal',  # Only this signal
    dbus_interface='org.test.Interface',
    path='/specific/path'  # Only this path
)

# BAD: Receive all signals and filter in handler
bus.add_signal_receiver(
    handler,
    signal_name=None,  # All signals - expensive!
    dbus_interface=None
)

パターン 3：dasbus による非同期呼び出し

# GOOD: Async calls for non-blocking operations
from dasbus.connection import SessionMessageBus
from dasbus.loop import EventLoop
import asyncio

async def async_call():
    bus = SessionMessageBus()
    proxy = bus.get_proxy('org.test.Service', '/')
    result = await asyncio.to_thread(proxy.Method)
    return result

# BAD: Blocking calls in async context
def blocking_call():
    bus = dbus.SessionBus()
    proxy = bus.get_object('org.test.Service', '/')
    return proxy.Method()  # Blocks event loop!

パターン 4：メッセージのバッチ処理

# GOOD: Batch property reads
def get_all_properties(proxy, interface):
    props = dbus.Interface(proxy, 'org.freedesktop.DBus.Properties')
    return props.GetAll(interface)  # One call

# BAD: Individual property reads
def get_properties_slow(proxy, interface):
    props = dbus.Interface(proxy, 'org.freedesktop.DBus.Properties')
    return {
        'prop1': props.Get(interface, 'prop1'),  # Call 1
        'prop2': props.Get(interface, 'prop2'),  # Call 2
        'prop3': props.Get(interface, 'prop3'),  # Call 3
    }

パターン 5：プロパティのキャッシュ

# GOOD: Cache properties with TTL
from functools import lru_cache
from time import time

class CachedPropertyAccess:
    def __init__(self, client, cache_ttl=5):
        self.client = client
        self.cache_ttl = cache_ttl
        self._cache = {}

    def get_property(self, bus_name, path, interface, prop):
        key = (bus_name, path, interface, prop)
        cached = self._cache.get(key)

        if cached and time() - cached['time'] < self.cache_ttl:
            return cached['value']

        value = self._fetch_property(bus_name, path, interface, prop)
        self._cache[key] = {'value': value, 'time': time()}
        return value

# BAD: Fetch property every time
def get_property(proxy, interface, prop):
    props = dbus.Interface(proxy, 'org.freedesktop.DBus.Properties')
    return props.Get(interface, prop)  # Always fetches

---

7. 実装パターン

パターン 1：セキュアな D-Bus クライアント

import dbus
from dbus.exceptions import DBusException
import logging

class SecureDBusClient:
    """Secure D-Bus client with access controls."""

    BLOCKED_SERVICES = {
        'org.freedesktop.PolicyKit1',          # Privilege escalation
        'org.freedesktop.systemd1',            # System service control
        'org.freedesktop.login1',              # Session/power management
        'org.gnome.keyring',                   # Secret storage
        'org.freedesktop.secrets',             # Secret service
        'org.freedesktop.PackageKit',          # Package installation
    }

    BLOCKED_INTERFACES = {
        'org.freedesktop.DBus.Properties',     # Can read/write any property
    }

    def __init__(self, bus_type: str = 'session', permission_tier: str = 'standard'):
        self.permission_tier = permission_tier
        self.logger = logging.getLogger('dbus.security')
        self.timeout = 30  # seconds

        # Connect to bus
        if bus_type == 'session':
            self.bus = dbus.SessionBus()
        elif bus_type == 'system':
            if permission_tier != 'elevated':
                raise PermissionError("System bus requires 'elevated' tier")
            self.bus = dbus.SystemBus()
        else:
            raise ValueError(f"Invalid bus type: {bus_type}")

    def get_object(self, bus_name: str, object_path: str) -> dbus.Interface:
        """Get D-Bus object with validation."""
        # Security check
        if bus_name in self.BLOCKED_SERVICES:
            self.logger.warning('blocked_service', service=bus_name)
            raise SecurityError(f"Access to {bus_name} is blocked")

        # Validate bus name format
        if not self._validate_bus_name(bus_name):
            raise ValueError(f"Invalid bus name: {bus_name}")

        # Get proxy object
        try:
            proxy = self.bus.get_object(bus_name, object_path)
            self._audit_log('get_object', bus_name, object_path)
            return proxy
        except DBusException as e:
            self.logger.error(f"D-Bus error: {e}")
            raise

    def call_method(
        self,
        bus_name: str,
        object_path: str,
        interface: str,
        method: str,
        *args
    ):
        """Call D-Bus method with validation."""
        # Security checks
        if interface in self.BLOCKED_INTERFACES:
            raise SecurityError(f"Interface {interface} is blocked")

        # Get object
        proxy = self.get_object(bus_name, object_path)
        iface = dbus.Interface(proxy, interface)

        # Call with timeout
        try:
            result = getattr(iface, method)(
                *args,
                timeout=self.timeout
            )
            self._audit_log('call_method', bus_name, f"{interface}.{method}")
            return result
        except DBusException as e:
            if 'Timeout' in str(e):
                raise TimeoutError(f"Method call timed out after {self.timeout}s")
            raise

    def get_peer_credentials(self, bus_name: str) -> dict:
        """Get credentials of D-Bus peer."""
        dbus_obj = self.bus.get_object(
            'org.freedesktop.DBus',
            '/org/freedesktop/DBus'
        )
        dbus_iface = dbus.Interface(dbus_obj, 'org.freedesktop.DBus')

        return {
            'pid': dbus_iface.GetConnectionUnixProcessID(bus_name),
            'uid': dbus_iface.GetConnectionUnixUser(bus_name),
        }

    def _validate_bus_name(self, name: str) -> bool:
        """Validate D-Bus bus name format."""
        import re
        pattern = r'^[a-zA-Z_][a-zA-Z0-9_]*(\.[a-zA-Z_][a-zA-Z0-9_]*)+$'
        return bool(re.match(pattern, name)) and len(name) <= 255

    def _audit_log(self, action: str, service: str, detail: str):
        """Log operation for audit."""
        self.logger.info(
            f'dbus.{action}',
            extra={
                'service': service,
                'detail': detail,
                'permission_tier': self.permission_tier
            }
        )

パターン 2：シグナルの監視

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

class SecureSignalMonitor:
    """Monitor D-Bus signals safely."""

    ALLOWED_SIGNALS = {
        'org.freedesktop.Notifications': ['NotificationClosed', 'ActionInvoked'],
        'org.freedesktop.FileManager1': ['OpenLocationRequested'],
    }

    def __init__(self, client: SecureDBusClient):
        self.client = client
        self.handlers = {}
        self.logger = logging.getLogger('dbus.signals')

        # Setup main loop
        DBusGMainLoop(set_as_default=True)

    def subscribe(
        self,
        bus_name: str,
        interface: str,
        signal: str,
        handler
    ):
        """Subscribe to signal with validation."""
        # Check if signal is allowed
        allowed = self.ALLOWED_SIGNALS.get(interface, [])
        if signal not in allowed:
            raise SecurityError(f"Signal {interface}.{signal} not allowed")

        # Wrapper to log signal receipt
        def safe_handler(*args):
            self.logger.info(
                'signal_received',
                extra={'interface': interface, 'signal': signal}
            )
            handler(*args)

        # Subscribe
        self.client.bus.add_signal_receiver(
            safe_handler,
            signal_name=signal,
            dbus_interface=interface,
            bus_name=bus_name
        )
        self.handlers[(interface, signal)] = safe_handler

    def run(self, timeout: int = None):
        """Run signal loop with timeout."""
        loop = GLib.MainLoop()

        if timeout:
            GLib.timeout_add_seconds(timeout, loop.quit)

        loop.run()

パターン 3：プロパティのアクセス制御

class SecurePropertyAccess:
    """Controlled access to D-Bus properties."""

    READABLE_PROPERTIES = {
        'org.freedesktop.Notifications': ['ServerCapabilities'],
        'org.mpris.MediaPlayer2': ['Identity', 'PlaybackStatus'],
    }

    WRITABLE_PROPERTIES = {
        'org.mpris.MediaPlayer2.Player': ['Volume'],
    }

    def __init__(self, client: SecureDBusClient):
        self.client = client
        self.logger = logging.getLogger('dbus.properties')

    def get_property(
        self,
        bus_name: str,
        object_path: str,
        interface: str,
        property_name: str
    ):
        """Get property with access control."""
        # Check if property is readable
        allowed = self.READABLE_PROPERTIES.get(interface, [])
        if property_name not in allowed:
            raise SecurityError(f"Property {interface}.{property_name} not readable")

        proxy = self.client.get_object(bus_name, object_path)
        props = dbus.Interface(proxy, 'org.freedesktop.DBus.Properties')

        value = props.Get(interface, property_name)
        self.logger.info(
            'property_read',
            extra={'interface': interface, 'property': property_name}
        )
        return value

    def set_property(
        self,
        bus_name: str,
        object_path: str,
        interface: str,
        property_name: str,
        value
    ):
        """Set property with access control."""
        if self.client.permission_tier == 'read-only':
            raise PermissionError("Setting properties requires 'standard' tier")

        # Check if property is writable
        allowed = self.WRITABLE_PROPERTIES.get(interface, [])
        if property_name not in allowed:
            raise SecurityError(f"Property {interface}.{property_name} not writable")

        proxy = self.client.get_object(bus_name, object_path)
        props = dbus.Interface(proxy, 'org.freedesktop.DBus.Properties')

        props.Set(interface, property_name, value)
        self.logger.info(
            'property_write',
            extra={'interface': interface, 'property': property_name}
        )

パターン 4：サービスの検出

class ServiceDiscovery:
    """Discover D-Bus services safely."""

    def __init__(self, client: SecureDBusClient):
        self.client = client

    def list_names(self) -> list:
        """List available bus names (filtered)."""
        dbus_obj = self.client.bus.get_object(
            'org.freedesktop.DBus',
            '/org/freedesktop/DBus'
        )
        dbus_iface = dbus.Interface(dbus_obj, 'org.freedesktop.DBus')

        all_names = dbus_iface.ListNames()

        # Filter blocked services
        filtered = [
            name for name in all_names
            if name not in SecureDBusClient.BLOCKED_SERVICES
        ]

        return filtered

    def introspect(self, bus_name: str, object_path: str) -> str:
        """Get introspection XML for object."""
        if bus_name in SecureDBusClient.BLOCKED_SERVICES:
            raise SecurityError(f"Cannot introspect {bus_name}")

        proxy = self.client.get_object(bus_name, object_path)
        return proxy.Introspect(
            dbus_interface='org.freedesktop.DBus.Introspectable'
        )

---

5. セキュリティ基準

5.1 重大な脆弱性

1. PolicyKit を介した権限昇格（CVE-2021-4034）

• 深刻度：CRITICAL
• 説明：ローカルでの権限昇格につながる Polkit の脆弱性
• 緩和策：PolicyKit サービスへのアクセスをブロックします

2. D-Bus 認証のバイパス（CVE-2022-42012）

• 深刻度：HIGH
• 説明：許可されていないセッションバスへのアクセス
• 緩和策：ピア認証情報を検証します

3. サービスのなりすまし（CWE-290）

• 深刻度：HIGH
• 説明：悪意のあるサービスが信頼された名前を名乗ること
• 緩和策：サービスの認証情報を検証します

4. メソッドインジェクション（CWE-74）

• 深刻度：MEDIUM
• 説明：悪意のあるメソッドパラメーター
• 緩和策：入力検証、サービス許可リスト

5. 情報漏えい（CWE-200）

• 重大度：中
• 説明：機密性の高いサービスデータの公開
• 軽減策：プロパティへのアクセス制御

5.2 権限レベルモデル

PERMISSION_TIERS = {
    'read-only': {
        'bus_type': 'session',
        'allowed_operations': ['get_property', 'introspect', 'list_names'],
        'blocked_services': BLOCKED_SERVICES,
    },
    'standard': {
        'bus_type': 'session',
        'allowed_operations': ['*', 'set_property', 'call_method'],
        'blocked_services': BLOCKED_SERVICES,
    },
    'elevated': {
        'bus_type': ['session', 'system'],
        'allowed_operations': ['*'],
        'blocked_services': ['org.freedesktop.PackageKit'],
    }
}

---

8. よくある間違い

禁止：必要性なくシステムバスにアクセスする

# BAD: Always use system bus
bus = dbus.SystemBus()

# GOOD: Prefer session bus
bus = dbus.SessionBus()
# Only use system bus when required

禁止：PolicyKit へのアクセスを許可する

# BAD: No service filtering
result = client.call_method('org.freedesktop.PolicyKit1', ...)

# GOOD: Block privileged services
if service not in BLOCKED_SERVICES:
    result = client.call_method(service, ...)

禁止：タイムアウトの適用を省略する

# BAD: No timeout
result = iface.SomeMethod()

# GOOD: With timeout
result = iface.SomeMethod(timeout=30)

---

13. デプロイ前チェックリスト

• サービスのブロックリストを設定済み
• システムバスよりセッションバスを優先
• すべての呼び出しにタイムアウトを適用
• ピア認証情報の検証
• 監査ログを有効化
• プロパティへのアクセス制御を設定済み

---

14. まとめ

目標は、以下を満たす D-Bus 自動化を作成することです。

• 安全性：サービスのブロックリスト、認証情報の検証、アクセス制御
• 信頼性：タイムアウトの適用、エラー処理
• 最小限：デフォルトでセッションバスを使用し、最小権限を適用

セキュリティに関する注意事項：

1. 常にシステムバスよりセッションバスを優先してください
2. PolicyKit と systemd へのアクセスをブロックしてください
3. 必要に応じてピア認証情報を検証してください
4. すべてのメソッド呼び出しにタイムアウトを適用してください
5. 監査のためにすべての操作を記録してください

---

参考資料

• references/security-examples.md を参照してください
• references/threat-model.md を参照してください
• references/advanced-patterns.md を参照してください