LinuxAT-SPI2自动化技能Skill linux-at-spi2

---

名称: linux-at-spi2 风险级别: 中等 描述: “在Linux桌面自动化中，专门于AT-SPI2（辅助技术服务提供商接口）的专家。通过D-Bus可访问性接口，专注于GTK/Qt应用程序的可访问性自动化。需要安全控制的高风险技能，以实现系统范围的访问。” 模型: sonnet

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. 验证目标应用程序身份
2. 检查阻止应用程序列表
3. 验证操作权限
4. 记录操作与关联ID
5. 强制执行超时限制

---

4. 技术基础

4.1 AT-SPI2架构

应用程序 -> ATK/QAccessible -> AT-SPI2注册表 -> D-Bus -> 客户端

关键组件:

• 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:
    """安全包装器用于AT-SPI2操作。"""

    BLOCKED_APPS = {
        'keepassxc', 'keepass2', 'bitwarden',  # 密码管理器
        'gnome-terminal', 'konsole', 'xterm',   # 终端
        'gnome-keyring', 'seahorse',            # 密钥管理
        'polkit-gnome-authentication-agent-1',  # 认证对话框
    }

    BLOCKED_ROLES = {
        Atspi.Role.PASSWORD_TEXT,  # 密码字段
    }

    def __init__(self, permission_tier: str = 'read-only'):
        self.permission_tier = permission_tier
        self.logger = logging.getLogger('atspi.security')
        self.timeout = 5000  # D-Bus调用的毫秒数

        # 初始化AT-SPI2
        Atspi.init()

    def get_desktop(self) -> 'Atspi.Accessible':
        """获取桌面根节点并超时。"""
        return Atspi.get_desktop(0)

    def get_application(self, name: str) -> 'Atspi.Accessible':
        """获取应用程序可访问对象并进行验证。"""
        name_lower = name.lower()

        # 安全检查
        if name_lower in self.BLOCKED_APPS:
            self.logger.warning('blocked_app', app=name)
            raise SecurityError(f"访问 {name} 被阻止")

        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:
        """获取对象值并进行安全过滤。"""
        # 检查密码字段
        if obj.get_role() in self.BLOCKED_ROLES:
            self.logger.warning('blocked_role', role=obj.get_role())
            raise SecurityError("访问密码字段被阻止")

        # 检查敏感名称
        name = obj.get_name().lower()
        if any(word in name for word in ['password', 'secret', 'token']):
            return '[已屏蔽]'

        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):
        """执行操作并进行权限检查。"""
        if self.permission_tier == 'read-only':
            raise PermissionError("操作需要'标准'层级")

        action = obj.get_action()
        if not action:
            raise ValueError("对象无操作")

        # 查找并执行操作
        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_name} 未找到")

    def _audit_log(self, event: str, detail: str):
        """记录操作用于审计。"""
        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("搜索超时")
            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:
    """安全监控AT-SPI2事件。"""
    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} 不允许")
        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("文本输入需要'标准'层级")
    if obj.get_role() == Atspi.Role.PASSWORD_TEXT:
        raise SecurityError("无法输入密码字段")

    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: 运行完整验证

# 运行所有测试并覆盖率
pytest tests/ -v --cov=automation --cov-report=term-missing

# 运行安全特定测试
pytest tests/ -k "security or blocked" -v

# 验证无密码字段访问
pytest tests/ -k "password" -v

---

7. 性能模式

模式1: 事件过滤（减少D-Bus流量）

# 坏: 注册所有事件
Atspi.EventListener.register_full(handler, 'object:', None)

# 好: 过滤到特定需要的事件
ALLOWED_EVENTS = ['object:state-changed:focused', 'window:activate']
for event in ALLOWED_EVENTS:
    Atspi.EventListener.register_full(handler, event, None)

模式2: 节点缓存（避免重复查找）

# 坏: 每个查询重新遍历树
def find_button():
    desktop = Atspi.get_desktop(0)
    for i in range(desktop.get_child_count()):
        app = desktop.get_child_at_index(i)
        # 每次完整树遍历

# 好: 缓存经常访问的节点
class CachedATSPI:
    def __init__(self):
        self._app_cache = {}
        self._cache_ttl = 5.0  # 秒

    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: 异步查询（非阻塞操作）

# 坏: 主线程中阻塞同步调用
buttons = [c for c in children if c.get_role() == PUSH_BUTTON]

# 好: 使用执行器进行重型树遍历
async def get_all_buttons_async(app):
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(None, lambda: find_buttons(app))

模式4: 连接池（单例）

# 坏: 每次操作调用Atspi.init()
# 好: 单例管理器
class ATSPIManager:
    _instance = None
    def __new__(cls):
        if not cls._instance:
            cls._instance = super().__new__(cls)
            Atspi.init()
        return cls._instance

模式5: 范围限制（减少搜索空间）

# 坏: 搜索整个桌面树
result = search_recursive(Atspi.get_desktop(0), name)

# 好: 限制到特定应用程序
app = get_application(app_name)
result = search_recursive(app, name)

# 更好: 添加角色过滤
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. 常见错误

绝不: 访问密码字段

# 坏: 无角色检查
value = obj.get_text().get_text(0, -1)

# 好: 先检查角色
if obj.get_role() != Atspi.Role.PASSWORD_TEXT:
    value = obj.get_text().get_text(0, -1)

绝不: 跳过应用程序验证

# 坏: 直接访问
app = desktop.get_child_at_index(0)
interact(app)

# 好: 先验证
if is_allowed_app(app.get_name()):
    interact(app)

---

10. 预实现清单

阶段1: 编写代码前

• [ ] 审查本技能中的AT-SPI2安全模式
• [ ] 识别目标应用程序并验证不在阻止列表中
• [ ] 确定所需权限层级（只读/标准/提升）
• [ ] 编写安全验证的失败测试
• [ ] 计划节点查找的缓存策略

阶段2: 实现期间

• [ ] 实现应用程序阻止列表检查
• [ ] 添加PASSWORD_TEXT角色阻止
• [ ] 在所有D-Bus调用上强制执行超时
• [ ] 应用节点缓存以提高性能
• [ ] 使用事件过滤（非通配符订阅）
• [ ] 实现搜索的范围限制

阶段3: 提交前

• [ ] 所有pytest测试通过，覆盖率>80%
• [ ] 验证所有操作的审计日志
• [ ] 在负载下测试速率限制
• [ ] 测试输出中无安全警告
• [ ] 验证性能（元素查找<100毫秒）

---

11. 总结

您的目标是创建AT-SPI2自动化，它应该是：

• 安全: 应用程序验证、角色阻止、审计日志
• 可靠: 超时强制执行、错误处理
• 可访问: 尊重辅助技术边界

安全提醒:

1. 总是阻止访问PASSWORD_TEXT角色
2. 在自动化前验证应用程序
3. 在所有D-Bus调用上强制执行超时
4. 记录所有操作用于审计
5. 使用适当的权限层级

---

参考文献

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