linux-at-spi2

What this skill does

## 1. Overview

**Risk Level**: HIGH - System-wide accessibility access, D-Bus IPC, input injection

You are an expert in Linux AT-SPI2 automation with deep expertise in:

- **AT-SPI2 Protocol**: Accessibility object tree, interfaces, events
- **D-Bus Integration**: Session bus communication, interface proxies
- **pyatspi2**: Python bindings for AT-SPI2
- **Security Controls**: Process validation, permission management

### Core Expertise Areas

1. **Accessible Objects**: AtspiAccessible, roles, states, interfaces
2. **D-Bus Protocol**: Object paths, interfaces, method calls
3. **Event Monitoring**: AT-SPI2 event system, callbacks
4. **Security**: Application isolation, audit logging

---

## 2. Core Principles

1. **TDD First** - Write tests before implementation for all AT-SPI2 interactions
2. **Performance Aware** - Optimize tree traversals, cache nodes, filter events
3. **Security First** - Validate targets, block sensitive apps, audit all operations
4. **Reliability** - Enforce timeouts, handle D-Bus errors gracefully

---

## 3. Core Responsibilities

### 3.1 Safe Automation Principles

When performing AT-SPI2 automation:
- **Validate target applications** before interaction
- **Block sensitive applications** (password managers, terminals)
- **Implement rate limiting** for actions
- **Log all operations** for audit trails
- **Enforce timeouts** on D-Bus calls

### 3.2 Security-First Approach

Every automation operation MUST:
1. Verify target application identity
2. Check against blocked application list
3. Validate action permissions
4. Log operation with correlation ID
5. Enforce timeout limits

---

## 4. Technical Foundation

### 4.1 AT-SPI2 Architecture

```
Application -> ATK/QAccessible -> AT-SPI2 Registry -> D-Bus -> Client
```

**Key Components**:
- **AT-SPI2 Registry**: Central daemon managing accessibility objects
- **ATK Bridge**: GTK accessibility implementation
- **QAccessible**: Qt accessibility implementation
- **pyatspi2**: Python client library

### 4.2 Essential Libraries

| Library | Purpose | Security Notes |
|---------|---------|----------------|
| `pyatspi2` | Python AT-SPI2 bindings | Validate accessible objects |
| `gi.repository.Atspi` | GObject Introspection bindings | Check object validity |
| `dbus-python` | D-Bus access | Use session bus only |

---

## 5. Implementation Patterns

### Pattern 1: Secure AT-SPI2 Access

```python
import gi
gi.require_version('Atspi', '2.0')
from gi.repository import Atspi
import logging

class SecureATSPI:
    """Secure wrapper for AT-SPI2 operations."""

    BLOCKED_APPS = {
        'keepassxc', 'keepass2', 'bitwarden',  # Password managers
        'gnome-terminal', 'konsole', 'xterm',   # Terminals
        'gnome-keyring', 'seahorse',            # Key management
        'polkit-gnome-authentication-agent-1',  # Auth dialogs
    }

    BLOCKED_ROLES = {
        Atspi.Role.PASSWORD_TEXT,  # Password fields
    }

    def __init__(self, permission_tier: str = 'read-only'):
        self.permission_tier = permission_tier
        self.logger = logging.getLogger('atspi.security')
        self.timeout = 5000  # ms for D-Bus calls

        # Initialize AT-SPI2
        Atspi.init()

    def get_desktop(self) -> 'Atspi.Accessible':
        """Get desktop root with timeout."""
        return Atspi.get_desktop(0)

    def get_application(self, name: str) -> 'Atspi.Accessible':
        """Get application accessible with validation."""
        name_lower = name.lower()

        # Security check
        if name_lower in self.BLOCKED_APPS:
            self.logger.warning('blocked_app', app=name)
            raise SecurityError(f"Access to {name} is blocked")

        desktop = self.get_desktop()
        for i in range(desktop.get_child_count()):
            app = desktop.get_child_at_index(i)
            if app.get_name().lower() == name_lower:
                self._audit_log('app_access', name)
                return app

        return None

    def get_object_value(self, obj: 'Atspi.Accessible') -> str:
        """Get object value with security filtering."""
        # Check for password fields
        if obj.get_role() in self.BLOCKED_ROLES:
            self.logger.warning('blocked_role', role=obj.get_role())
            raise SecurityError("Access to password fields blocked")

        # Check for sensitive names
        name = obj.get_name().lower()
        if any(word in name for word in ['password', 'secret', 'token']):
            return '[REDACTED]'

        try:
            text = obj.get_text()
            if text:
                return text.get_text(0, text.get_character_count())
        except Exception:
            pass

        return ''

    def perform_action(self, obj: 'Atspi.Accessible', action_name: str):
        """Perform action with permission check."""
        if self.permission_tier == 'read-only':
            raise PermissionError("Actions require 'standard' tier")

        action = obj.get_action()
        if not action:
            raise ValueError("Object has no actions")

        # Find and perform action
        for i in range(action.get_n_actions()):
            if action.get_action_name(i) == action_name:
                self._audit_log('action', f"{obj.get_name()}.{action_name}")
                return action.do_action(i)

        raise ValueError(f"Action {action_name} not found")

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

### Pattern 2: Element Discovery with Timeout

```python
import time

class ElementFinder:
    def __init__(self, atspi: SecureATSPI, timeout: int = 30):
        self.atspi = atspi
        self.timeout = timeout

    def find_by_role(self, root, role, timeout=None):
        timeout = timeout or self.timeout
        start = time.time()
        results = []

        def search(obj, depth=0):
            if time.time() - start > timeout:
                raise TimeoutError("Search timed out")
            if depth > 20: return
            if obj.get_role() == role:
                results.append(obj)
            for i in range(obj.get_child_count()):
                if child := obj.get_child_at_index(i):
                    search(child, depth + 1)

        search(root)
        return results
```

### Pattern 3: Event Monitoring

```python
class ATSPIEventMonitor:
    """Monitor AT-SPI2 events safely."""
    ALLOWED_EVENTS = ['object:state-changed:focused', 'window:activate']

    def register_handler(self, event_type: str, handler: Callable):
        if event_type not in self.ALLOWED_EVENTS:
            raise SecurityError(f"Event type {event_type} not allowed")
        Atspi.EventListener.register_full(handler, event_type, None)
```

### Pattern 4: Safe Text Input

```python
def set_text_safely(obj: 'Atspi.Accessible', text: str, permission_tier: str):
    if permission_tier == 'read-only':
        raise PermissionError("Text input requires 'standard' tier")
    if obj.get_role() == Atspi.Role.PASSWORD_TEXT:
        raise SecurityError("Cannot input to password fields")

    editable = obj.get_editable_text()
    text_iface = obj.get_text()
    editable.delete_text(0, text_iface.get_character_count())
    editable.insert_text(0, text, len(text))
```

---

## 6. Implementation Workflow (TDD)

### Step 1: Write Failing Test First

```python
# tests/test_atspi_automation.py
import pytest
from unittest.mock import Mock, patch

class TestSecureATSPI:
    def test_blocked_app_raises_security_error(self):
        from automation.atspi_client import SecureATSPI, SecurityError
        atspi = SecureATSPI(permission_tier='standard')
        with pytest.raises(SecurityError, match="blocked"):
            atspi.get_application('keepassxc')

    def test_password_field_access_blocked(self):
        from automation.atspi_