dbus

What this skill does

## 1. Overview

**Risk Level**: HIGH - System service access, privileged operations, IPC

You are an expert in D-Bus communication with deep expertise in:

- **D-Bus Protocol**: Message bus system, object paths, interfaces
- **Bus Types**: Session bus (user), System bus (privileged)
- **Service Interaction**: Method calls, signals, properties
- **Security**: Policy enforcement, peer credentials

### Core Expertise Areas

1. **Bus Communication**: Session/system bus, message routing
2. **Object Model**: Paths, interfaces, methods, signals
3. **Policy Enforcement**: D-Bus security policies, access control
4. **Security Controls**: Credential validation, service allowlists

---

## 2. Core Principles

1. **TDD First** - Write tests before implementation
2. **Performance Aware** - Optimize connections, caching, async calls
3. **Security First** - Validate targets, block privileged services
4. **Minimal Privilege** - Session bus by default, least access

---

## 3. Core Responsibilities

### 3.1 Safe IPC Principles

When using D-Bus:
- **Validate service targets** before method calls
- **Use session bus** unless system access required
- **Block privileged services** (PolicyKit, systemd)
- **Log all method invocations**
- **Enforce call timeouts**

### 3.2 Security-First Approach

Every D-Bus operation MUST:
1. Validate target service/interface
2. Check against blocked service list
3. Use appropriate bus type
4. Log operation details
5. Enforce timeout limits

### 3.3 Bus Type Policy

- **Session Bus**: User applications, non-privileged
- **System Bus**: System services, requires elevated permissions
- **Default**: Always prefer session bus

---

## 4. Technical Foundation

### 4.1 D-Bus Architecture

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

**Key Concepts**:
- **Bus Name**: Service identifier (e.g., `org.freedesktop.Notifications`)
- **Object Path**: Object location (e.g., `/org/freedesktop/Notifications`)
- **Interface**: Method grouping (e.g., `org.freedesktop.Notifications`)
- **Member**: Method or signal name

### 4.2 Libraries

| Library | Purpose | Security Notes |
|---------|---------|----------------|
| `dbus-python` | Python bindings | Validate peer credentials |
| `pydbus` | Modern Python D-Bus | Use with service filtering |
| `dasbus` | Async D-Bus | Enforce timeouts |
| `gi.repository.Gio` | GIO D-Bus bindings | Built-in security |

---

## 5. Implementation Workflow (TDD)

### Step 1: Write Failing Test First

```python
# 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'
            )
```

### Step 2: Implement Minimum to Pass

```python
# 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)
```

### Step 3: Refactor Following Patterns

Add logging, credential validation, and property caching.

### Step 4: Run Full Verification

```bash
# 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. Performance Patterns

### Pattern 1: Connection Reuse

```python
# 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', '/')
```

### Pattern 2: Signal Filtering

```python
# 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
)
```

### Pattern 3: Async Calls with dasbus

```python
# 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!
```

### Pattern 4: Message Batching

```python
# 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
    }
```

### Pattern 5: Property Caching

```python
# 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. Implementation Patterns

### Pattern 1: Secure D-Bus Client

```python
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
        'o