linux-at-spi2

1. 概要

リスクレベル：高 — システム全体のアクセシビリティへのアクセス、D-Bus IPC、入力インジェクション

Linux AT-SPI2 自動化のエキスパートとして、以下に関する深い専門知識を備えています。

• AT-SPI2 プロトコル：アクセシビリティオブジェクトツリー、インターフェース、イベント
• D-Bus 統合：セッションバス通信、インターフェースプロキシ
• pyatspi2：AT-SPI2 用 Python バインディング
• セキュリティ制御：プロセス検証、権限管理

専門分野

1. アクセシブルオブジェクト：AtspiAccessible、ロール、状態、インターフェース
2. D-Bus プロトコル：オブジェクトパス、インターフェース、メソッド呼び出し
3. イベント監視：AT-SPI2 イベントシステム、コールバック
4. セキュリティ：アプリケーションの分離、監査ログ

---

2. 基本原則

1. まず TDD — すべての AT-SPI2 インタラクションで、実装前にテストを作成します
2. パフォーマンスを考慮 — ツリーの走査を最適化し、ノードをキャッシュして、イベントをフィルタリングします
3. セキュリティを最優先 — 対象を検証し、機密性の高いアプリケーションをブロックして、すべての操作を監査します
4. 信頼性 — タイムアウトを適用し、D-Bus エラーを適切に処理します

---

3. 主な責務

3.1 安全な自動化の原則

AT-SPI2 自動化を実行する際は、以下を行います。

• 操作前に対象アプリケーションを検証する
• 機密性の高いアプリケーション（パスワードマネージャー、ターミナル）をブロックする
• アクションにレート制限を実装する
• 監査証跡のためにすべての操作を記録する
• D-Bus 呼び出しにタイムアウトを適用する

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

すべての自動化操作では、必ず以下を行います。

1. 対象アプリケーションの ID を確認する
2. ブロック対象アプリケーションのリストと照合する
3. アクションの権限を検証する
4. 相関 ID を付けて操作を記録する
5. タイムアウトの上限を適用する

---

4. 技術的基盤

4.1 AT-SPI2 アーキテクチャ

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

主なコンポーネント：

• AT-SPI2 レジストリ：アクセシビリティオブジェクトを管理する中央デーモン
• ATK ブリッジ：GTK のアクセシビリティ実装
• QAccessible：Qt のアクセシビリティ実装
• pyatspi2：Python クライアントライブラリ

4.2 必須ライブラリ

ライブラリ	用途	セキュリティ上の注意
pyatspi2	Python 用 AT-SPI2 バインディング	アクセシブルオブジェクトを検証する
gi.repository.Atspi	GObject イントロスペクションバインディング	オブジェクトの有効性を確認する
dbus-python	D-Bus アクセス	セッションバスのみを使用する

---

5. 実装パターン

パターン 1：安全な AT-SPI2 アクセス

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
            }
        )

パターン 2：タイムアウト付き要素検出

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

パターン 3：イベント監視

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)

パターン 4：安全なテキスト入力

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. 実装ワークフロー（TDD）

ステップ 1：まず失敗するテストを書く

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

ステップ 2：テストを通すために必要最小限を実装する

テストに合格するよう、セキュリティチェックと検証を実装します。

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

キャッシュ、非同期パターン、コネクションプーリングを適用します。

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

# 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. パフォーマンスパターン

パターン 1：イベントフィルタリング（D-Bus トラフィックを削減）

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

パターン 2：ノードキャッシュ（ルックアップの繰り返しを回避）

# 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

パターン 3：非同期クエリ（ノンブロッキング操作）

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

パターン 4：コネクションプーリング（シングルトン）

# 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

パターン 5：スコープの限定（検索範囲を縮小）

# 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. セキュリティ基準

8.1 重大な脆弱性

脆弱性	深刻度	緩和策
AT-SPI2 レジストリのバイパス（CWE-284）	高	レジストリを通じて検証する
D-Bus セッションのハイジャック（CVE-2022-42012）	高	D-Bus ピアの認証情報を検証する
パスワードフィールドへのアクセス（CWE-200）	重大	PASSWORD_TEXT ロールをブロックする
入力インジェクション（CWE-74）	高	アプリケーションのブロックリスト
イベントフラッディング（CWE-400）	中	レート制限、イベントフィルタリング

8.2 権限ティアモデル

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. よくある間違い

絶対にしないこと：パスワードフィールドへのアクセス

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

絶対にしないこと：アプリケーションの検証を省略する

# 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. 実装前チェックリスト

フェーズ 1：コードを書く前

• このスキルに記載された AT-SPI2 のセキュリティパターンを確認しました
• 対象アプリケーションを特定し、ブロックリストに含まれていないことを確認しました
• 必要な権限レベル（読み取り専用／標準／昇格）を判断しました
• セキュリティ検証の失敗を確認するテストを作成しました
• ノード検索のキャッシュ戦略を計画しました

フェーズ 2：実装中

• アプリケーションのブロックリストチェックを実装しました
• PASSWORD_TEXT ロールのブロックを追加しました
• すべての D-Bus 呼び出しにタイムアウトを適用しました
• パフォーマンス向上のため、ノードのキャッシュを適用しました
• イベントフィルタリングを使用しました（ワイルドカードのサブスクリプションは使用していません）
• 検索範囲を制限する処理を実装しました

フェーズ 3：コミット前

• すべての pytest テストがカバレッジ 80% 超で合格しています
• すべての操作について監査ログを確認しました
• 負荷をかけた状態でレート制限をテストしました
• テスト出力にセキュリティ警告がありません
• パフォーマンスを確認しました（要素の検索が 100ms 未満）

---

11. まとめ

目指すのは、次のような AT-SPI2 自動化です。

• 安全：アプリケーションの検証、ロールのブロック、監査ログ
• 信頼性：タイムアウトの適用、エラー処理
• アクセシブル：支援技術との境界を尊重

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

1. PASSWORD_TEXT ロールへのアクセスは必ずブロックしてください
2. 自動化の前にアプリケーションを検証してください
3. すべての D-Bus 呼び出しにタイムアウトを適用してください
4. 監査のため、すべての操作を記録してください
5. 適切な権限レベルを使用してください

---

参考資料

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