Linux At Spi2

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

---

5. Implementation Patterns

Pattern 1: Secure AT-SPI2 Access

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

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

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

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

# 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_client import SecureATSPI, SecurityError
        atspi = SecureATSPI()
        mock_obj = Mock()
        mock_obj.get_role.return_value = 24  # PASSWORD_TEXT
        with pytest.raises(SecurityError):
            atspi.get_object_value(mock_obj)

    def test_read_only_tier_blocks_actions(self):
        from automation.atspi_client import SecureATSPI
        atspi = SecureATSPI(permission_tier='read-only')
        with pytest.raises(PermissionError):
            atspi.perform_action(Mock(), 'click')

Step 2: Implement Minimum to Pass

Implement the security checks and validations to pass tests.

Step 3: Refactor Following Patterns

Apply caching, async patterns, and connection pooling.

Step 4: Run Full Verification

# Run all tests with coverage
pytest tests/ -v --cov=automation --cov-report=term-missing

# Run security-specific tests
pytest tests/ -k "security or blocked" -v

# Verify no password field access
pytest tests/ -k "password" -v

---

7. Performance Patterns

Pattern 1: Event Filtering (Reduce D-Bus Traffic)

# BAD: Register for all events
Atspi.EventListener.register_full(handler, 'object:', None)

# GOOD: Filter to specific events needed
ALLOWED_EVENTS = ['object:state-changed:focused', 'window:activate']
for event in ALLOWED_EVENTS:
    Atspi.EventListener.register_full(handler, event, None)

Pattern 2: Node Caching (Avoid Repeated Lookups)

# BAD: Re-traverse tree for each query
def find_button():
    desktop = Atspi.get_desktop(0)
    for i in range(desktop.get_child_count()):
        app = desktop.get_child_at_index(i)
        # Full tree traversal every time

# GOOD: Cache frequently accessed nodes
class CachedATSPI:
    def __init__(self):
        self._app_cache = {}
        self._cache_ttl = 5.0  # seconds

    def get_application(self, name: str):
        now = time.time()
        if name in self._app_cache:
            cached, timestamp = self._app_cache[name]
            if now - timestamp < self._cache_ttl:
                return cached

        app = self._find_app(name)
        self._app_cache[name] = (app, now)
        return app

Pattern 3: Async Queries (Non-Blocking Operations)

# BAD: Blocking synchronous calls in main thread
buttons = [c for c in children if c.get_role() == PUSH_BUTTON]

# GOOD: Use executor for heavy tree traversals
async def get_all_buttons_async(app):
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(None, lambda: find_buttons(app))

Pattern 4: Connection Pooling (Singleton)

# BAD: Atspi.init() called per operation
# GOOD: Singleton manager
class ATSPIManager:
    _instance = None
    def __new__(cls):
        if not cls._instance:
            cls._instance = super().__new__(cls)
            Atspi.init()
        return cls._instance

Pattern 5: Scope Limiting (Reduce Search Space)

# BAD: Search entire desktop tree
result = search_recursive(Atspi.get_desktop(0), name)

# GOOD: Limit to specific app
app = get_application(app_name)
result = search_recursive(app, name)

# BETTER: Add role filtering
result = search_with_role(app, name, role=Atspi.Role.PUSH_BUTTON)

---

8. Security Standards

8.1 Critical Vulnerabilities

8.2 Permission Tier Model

PERMISSION_TIERS = {
    'read-only': {
        'allowed_operations': ['get_name', 'get_role', 'get_state', 'find'],
        'blocked_roles': [Atspi.Role.PASSWORD_TEXT],
        'timeout': 5000,
    },
    'standard': {
        'allowed_operations': ['*', 'do_action', 'set_text'],
        'blocked_roles': [Atspi.Role.PASSWORD_TEXT],
        'timeout': 10000,
    },
    'elevated': {
        'allowed_operations': ['*'],
        'blocked_apps': ['polkit', 'gnome-keyring'],
        'timeout': 30000,
    }
}

---

9. Common Mistakes

Never: Access Password Fields

# BAD: No role check
value = obj.get_text().get_text(0, -1)

# GOOD: Check role first
if obj.get_role() != Atspi.Role.PASSWORD_TEXT:
    value = obj.get_text().get_text(0, -1)

Never: Skip Application Validation

# BAD: Direct access
app = desktop.get_child_at_index(0)
interact(app)

# GOOD: Validate first
if is_allowed_app(app.get_name()):
    interact(app)

---

10. Pre-Implementation Checklist

Phase 1: Before Writing Code

• [ ] Reviewed AT-SPI2 security patterns in this skill
• [ ] Identified target applications and verified not in blocklist
• [ ] Determined required permission tier (read-only/standard/elevated)
• [ ] Wrote failing tests for security validations
• [ ] Planned caching strategy for node lookups

Phase 2: During Implementation

• [ ] Implemented application blocklist checks
• [ ] Added PASSWORD_TEXT role blocking
• [ ] Enforced timeouts on all D-Bus calls
• [ ] Applied node caching for performance
• [ ] Used event filtering (not wildcard subscriptions)
• [ ] Implemented scope limiting for searches

Phase 3: Before Committing

• [ ] All pytest tests pass with coverage > 80%
• [ ] Audit logging verified for all operations
• [ ] Rate limiting tested under load
• [ ] No security warnings in test output
• [ ] Performance verified (< 100ms for element lookups)

---

11. Summary

Your goal is to create AT-SPI2 automation that is:

• Secure: Application validation, role blocking, audit logging
• Reliable: Timeout enforcement, error handling
• Accessible: Respects assistive technology boundaries

Security Reminders: 1. Always block access to PASSWORD_TEXT roles 2. Validate applications before automation 3. Enforce timeouts on all D-Bus calls 4. Log all operations for audit 5. Use appropriate permission tiers

---

References

• See references/security-examples.md
• See references/threat-model.md
• See references/advanced-patterns.md

AT-SPI2 - Advanced Patterns

Pattern: Application Session Manager

class ATSPISession:
    """Managed AT-SPI2 session with cleanup."""

    def __init__(self, permission_tier: str = 'read-only'):
        self.permission_tier = permission_tier
        self.atspi = SecureATSPI(permission_tier)
        self.event_monitor = ATSPIEventMonitor()
        self.active = False

    def __enter__(self):
        Atspi.init()
        self.active = True
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.event_monitor.deregister_all()
        Atspi.exit()
        self.active = False
        return False

Pattern: Cached Object Tree

class CachedObjectTree:
    """Cache AT-SPI2 object tree with TTL."""

    def __init__(self, ttl: int = 5):
        self.ttl = ttl
        self.cache = {}
        self.timestamps = {}

    def get_children(self, obj: Atspi.Accessible) -> list:
        """Get children with caching."""
        obj_hash = self._get_hash(obj)
        now = time.time()

        if obj_hash in self.cache:
            if now - self.timestamps[obj_hash] < self.ttl:
                return self.cache[obj_hash]

        children = [
            obj.get_child_at_index(i)
            for i in range(obj.get_child_count())
        ]

        self.cache[obj_hash] = children
        self.timestamps[obj_hash] = now
        return children

    def invalidate(self):
        """Clear cache."""
        self.cache.clear()
        self.timestamps.clear()

Pattern: Async Event Processing

import asyncio
from gi.repository import GLib

class AsyncATSPIEvents:
    """Process AT-SPI2 events asynchronously."""

    def __init__(self):
        self.queue = asyncio.Queue()
        self.running = False

    def start(self):
        """Start event loop."""
        self.running = True
        loop = GLib.MainLoop()

        # Run in thread
        import threading
        thread = threading.Thread(target=loop.run)
        thread.daemon = True
        thread.start()

    async def get_event(self, timeout: float = None):
        """Get next event from queue."""
        try:
            return await asyncio.wait_for(
                self.queue.get(),
                timeout=timeout
            )
        except asyncio.TimeoutError:
            return None

Pattern: State Machine for Automation

from enum import Enum, auto

class AutomationState(Enum):
    IDLE = auto()
    SEARCHING = auto()
    INTERACTING = auto()
    COMPLETED = auto()
    ERROR = auto()

class AutomationStateMachine:
    """State machine for AT-SPI2 automation."""

    def __init__(self):
        self.state = AutomationState.IDLE
        self.logger = logging.getLogger('atspi.state')

    def transition(self, new_state: AutomationState):
        """Transition to new state."""
        old_state = self.state
        self.state = new_state
        self.logger.info(
            'state_transition',
            extra={'from': old_state.name, 'to': new_state.name}
        )

    def can_interact(self) -> bool:
        """Check if interaction is allowed."""
        return self.state in [
            AutomationState.IDLE,
            AutomationState.SEARCHING
        ]

Pattern: Object Path Tracking

class ObjectPathTracker:
    """Track path to AT-SPI2 objects for replay."""

    def __init__(self):
        self.paths = {}

    def record_path(self, obj: Atspi.Accessible) -> list:
        """Record path from root to object."""
        path = []
        current = obj

        while current:
            path.append({
                'role': current.get_role(),
                'name': current.get_name(),
                'index': self._get_index(current)
            })
            current = current.get_parent()

        path.reverse()
        return path

    def replay_path(self, path: list) -> Atspi.Accessible:
        """Navigate to object using recorded path."""
        current = Atspi.get_desktop(0)

        for step in path[1:]:  # Skip desktop
            found = False
            for i in range(current.get_child_count()):
                child = current.get_child_at_index(i)
                if (child.get_role() == step['role'] and
                    child.get_name() == step['name']):
                    current = child
                    found = True
                    break

            if not found:
                raise ElementNotFoundError(f"Could not find: {step}")

        return current

AT-SPI2 - Security Examples

Role-Based Access Control

BLOCKED_ROLES = {
    Atspi.Role.PASSWORD_TEXT: 'Password input',
    Atspi.Role.TERMINAL: 'Terminal emulator',
}

SENSITIVE_ROLES = {
    Atspi.Role.TEXT: ['password', 'secret', 'token', 'key'],
}

def check_role_access(obj: Atspi.Accessible) -> bool:
    """Validate access based on object role."""
    role = obj.get_role()

    if role in BLOCKED_ROLES:
        return False

    if role in SENSITIVE_ROLES:
        name = obj.get_name().lower()
        if any(word in name for word in SENSITIVE_ROLES[role]):
            return False

    return True

D-Bus Credential Validation

import dbus

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

    pid = dbus_iface.GetConnectionUnixProcessID(bus_name)
    uid = dbus_iface.GetConnectionUnixUser(bus_name)

    return {'pid': pid, 'uid': uid, 'bus_name': bus_name}

Audit Logging

import json
import logging

class ATSPIAuditLogger:
    """Comprehensive audit logging."""

    def log_operation(self, operation: str, app: str, element: str, success: bool):
        record = {
            'timestamp': datetime.utcnow().isoformat(),
            'event': 'atspi_operation',
            'operation': operation,
            'application': app,
            'element': element,
            'success': success
        }
        logging.getLogger('atspi.audit').info(json.dumps(record))

    def log_blocked_access(self, reason: str, app: str, role: str):
        record = {
            'timestamp': datetime.utcnow().isoformat(),
            'event': 'atspi_blocked',
            'reason': reason,
            'application': app,
            'role': role
        }
        logging.getLogger('atspi.audit').warning(json.dumps(record))

Rate Limiting

import time
from collections import defaultdict

class ActionRateLimiter:
    """Rate limit AT-SPI2 actions."""

    def __init__(self, max_actions: int = 30, period: int = 60):
        self.max_actions = max_actions
        self.period = period
        self.actions = defaultdict(list)

    def check(self, app_name: str) -> bool:
        """Check if action is allowed."""
        now = time.time()
        cutoff = now - self.period

        # Clean old entries
        self.actions[app_name] = [
            t for t in self.actions[app_name] if t > cutoff
        ]

        # Check limit
        if len(self.actions[app_name]) >= self.max_actions:
            return False

        self.actions[app_name].append(now)
        return True

Input Validation

def validate_search_criteria(criteria: dict) -> bool:
    """Validate AT-SPI2 search criteria."""
    allowed_keys = {'name', 'role', 'state', 'description'}

    for key in criteria:
        if key not in allowed_keys:
            return False

    if 'name' in criteria:
        if len(criteria['name']) > 255:
            return False
        if not criteria['name'].isprintable():
            return False

    return True

AT-SPI2 - Threat Model

Threat Model Overview

Domain Risk Level: HIGH Attack Surface: System-wide accessibility, D-Bus session bus

Assets to Protect

1. User Credentials - CRITICAL - Password fields, key managers 2. System Integrity - HIGH - Input injection prevention 3. User Privacy - HIGH - Screen content, application data

---

Attack Scenario 1: Password Field Harvesting

Threat Level: CRITICAL

Attack Flow:

1. Enumerate all accessible objects
2. Find PASSWORD_TEXT role objects
3. Read text content
4. Exfiltrate credentials

Mitigation: Block all access to PASSWORD_TEXT role

---

Attack Scenario 2: Input Injection to Terminals

Threat Level: CRITICAL

Attack Flow:

1. Find terminal emulator accessible
2. Use EditableText interface
3. Inject malicious commands
4. Execute arbitrary code

Mitigation: Block terminal applications in automation

---

Attack Scenario 3: Keyring Access

Threat Level: CRITICAL

Attack Flow:

1. Target gnome-keyring or seahorse
2. Enumerate secret entries
3. Extract stored passwords
4. Exfiltrate secrets

Mitigation: Block keyring applications

---

Attack Scenario 4: D-Bus Session Bus Exploitation

Threat Level: HIGH

Attack Flow:

1. Connect to session bus
2. Enumerate AT-SPI2 objects
3. Access sensitive applications
4. Extract or inject data

Mitigation: Validate D-Bus peer credentials, application filtering

---

STRIDE Analysis

---

Security Controls

Preventive

• PASSWORD_TEXT role blocking
• Application blocklists
• D-Bus peer validation
• Permission tier enforcement

Detective

• Comprehensive audit logging
• D-Bus activity monitoring
• Anomaly detection

Corrective

• Automatic rate limiting
• Session termination on violations