macos-accessibility

ApplicationServices.framework  # Core accessibility APIs
CoreFoundation.framework       # CFType support
AppKit.framework              # NSRunningApplication
Security.framework            # TCC queries

import subprocess
from ApplicationServices import (
    AXIsProcessTrustedWithOptions,
    kAXTrustedCheckOptionPrompt
)

class TCCValidator:
    """Validate TCC permissions before automation."""

    @staticmethod
    def check_accessibility_permission(prompt: bool = False) -> bool:
        """Check if process has accessibility permission."""
        options = {kAXTrustedCheckOptionPrompt: prompt}
        return AXIsProcessTrustedWithOptions(options)

    @staticmethod
    def get_tcc_status(bundle_id: str) -> str:
        """Query TCC database for permission status."""
        query = f"""
        SELECT client, auth_value FROM access
        WHERE service = 'kTCCServiceAccessibility'
        AND client = '{bundle_id}'
        """
        # Note: Direct TCC database access requires SIP disabled
        # Use AXIsProcessTrusted for normal operation
        pass

    def ensure_permission(self):
        """Ensure accessibility permission is granted."""
        if not self.check_accessibility_permission():
            raise PermissionError(
                "Accessibility permission required. "
                "Enable in System Preferences > Security & Privacy > Accessibility"
            )

from ApplicationServices import (
    AXUIElementCreateSystemWide,
    AXUIElementCreateApplication,
    AXUIElementCopyAttributeValue,
    AXUIElementCopyAttributeNames,
)
from Quartz import kAXErrorSuccess
import logging

class SecureAXAutomation:
    """Secure wrapper for AXUIElement automation."""

    BLOCKED_APPS = {
        'com.apple.keychainaccess',           # Keychain Access
        'com.apple.systempreferences',         # System Preferences
        'com.apple.SecurityAgent',             # Security dialogs
        'com.apple.Terminal',                  # Terminal
        'com.1password.1password',             # 1Password
    }

    def __init__(self, permission_tier: str = 'read-only'):
        self.permission_tier = permission_tier
        self.logger = logging.getLogger('ax.security')
        self.operation_timeout = 30

        # Validate TCC permission on init
        if not TCCValidator.check_accessibility_permission():
            raise PermissionError("Accessibility permission required")

    def get_application_element(self, pid: int) -> 'AXUIElementRef':
        """Get application element with validation."""
        # Get bundle ID
        bundle_id = self._get_bundle_id(pid)

        # Security check
        if bundle_id in self.BLOCKED_APPS:
            self.logger.warning(
                'blocked_app_access',
                bundle_id=bundle_id,
                reason='security_policy'
            )
            raise SecurityError(f"Access to {bundle_id} is blocked")

        # Create element
        app_element = AXUIElementCreateApplication(pid)

        self._audit_log('app_element_created', bundle_id, pid)
        return app_element

    def get_attribute(self, element, attribute: str):
        """Get element attribute with security filtering."""
        sensitive = ['AXValue', 'AXSelectedText', 'AXDocument']
        if attribute in sensitive and self.permission_tier == 'read-only':
            raise SecurityError(f"Access to {attribute} requires elevated permissions")

        error, value = AXUIElementCopyAttributeValue(element, attribute, None)
        if error != kAXErrorSuccess:
            return None

        # Redact password values
        return '[REDACTED]' if 'password' in str(attribute).lower() else value

    def _audit_log(self, action: str, bundle_id: str, pid: int):
        self.logger.info(f'ax.{action}', extra={
            'bundle_id': bundle_id, 'pid': pid, 'permission_tier': self.permission_tier
        })

from ApplicationServices import AXUIElementPerformAction

class SafeActionExecutor:
    """Execute AX actions with security controls."""
    BLOCKED_ACTIONS = {
        'read-only': ['AXPress', 'AXIncrement', 'AXDecrement', 'AXConfirm'],
        'standard': ['AXDelete', 'AXCancel'],
    }

    def __init__(self, permission_tier: str):
        self.permission_tier = permission_tier

    def perform_action(self, element, action: str):
        blocked = self.BLOCKED_ACTIONS.get(self.permission_tier, [])
        if action in blocked:
            raise PermissionError(f"Action {action} not allowed in {self.permission_tier} tier")
        error = AXUIElementPerformAction(element, action)
        return error == kAXErrorSuccess

from AppKit import NSWorkspace, NSRunningApplication

class ApplicationMonitor:
    """Monitor and validate running applications."""

    def get_frontmost_app(self) -> dict:
        app = NSWorkspace.sharedWorkspace().frontmostApplication()
        return {
            'pid': app.processIdentifier(),
            'bundle_id': app.bundleIdentifier(),
            'name': app.localizedName(),
        }

    def validate_application(self, pid: int) -> bool:
        app = NSRunningApplication.runningApplicationWithProcessIdentifier_(pid)
        if not app or app.bundleIdentifier() in SecureAXAutomation.BLOCKED_APPS:
            return False
        # Verify code signature
        result = subprocess.run(['codesign', '-v', app.bundleURL().path()], capture_output=True)
        return result.returncode == 0

# tests/test_ax_automation.py
import pytest
from unittest.mock import patch, MagicMock

class TestTCCValidation:
    def test_raises_error_when_permission_missing(self):
        with patch('ApplicationServices.AXIsProcessTrustedWithOptions', return_value=False):
            with pytest.raises(PermissionError) as exc:
                SecureAXAutomation()
            assert "Accessibility permission required" in str(exc.value)

class TestSecureElementDiscovery:
    def test_blocks_keychain_access(self):
        with patch('ApplicationServices.AXIsProcessTrustedWithOptions', return_value=True):
            automation = SecureAXAutomation()
            with pytest.raises(SecurityError):
                automation.get_application_element(pid=1234)  # Keychain PID

    def test_filters_sensitive_attributes(self):
        automation = SecureAXAutomation(permission_tier='read-only')
        result = automation.get_attribute(MagicMock(), 'AXPasswordField')
        assert result == '[REDACTED]'

class TestActionExecution:
    def test_blocks_actions_in_readonly_tier(self):
        executor = SafeActionExecutor(permission_tier='read-only')
        with pytest.raises(PermissionError):
            executor.perform_action(MagicMock(), 'AXPress')

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

# Run security-specific tests
pytest tests/test_ax_automation.py -k "security or permission" -v

# Run with timeout to catch hangs
pytest tests/ --timeout=30

# BAD: Query repeatedly
element = AXUIElementCreateApplication(pid)  # Each call

# GOOD: Cache with TTL
class ElementCache:
    def __init__(self, ttl=5.0):
        self.cache, self.ttl = {}, ttl

    def get_or_create(self, pid, role):
        key = (pid, role)
        if key in self.cache and time() - self.cache[key][1] < self.ttl:
            return self.cache[key][0]
        element = self._create_element(pid, role)
        self.cache[key] = (element, time())
        return element

# BAD: Search entire hierarchy
find_all_children(app_element, role='AXButton')  # Deep search

# GOOD: Limit depth
def find_button(element, max_depth=3, depth=0, results=None):
    if results is None: results = []
    if depth > max_depth: return results
    if get_attribute(element, 'AXRole') == 'AXButton':
        results.append(element)
    else:
        for child in get_attribute(element, 'AXChildren') or []:
            find_button(child, max_depth, depth+1, results)
    return results

# BAD: Sequential blocking
for app in apps: windows.extend(get_windows(app))

# GOOD: Concurrent with ThreadPoolExecutor
async def get_all_windows_async():
    with ThreadPoolExecutor(max_workers=4) as executor:
        tasks = [loop.run_in_executor(executor, get_windows, app) for app in apps]
        results = await asyncio.gather(*tasks)
    return [w for wins in results for w in wins]

# BAD: Multiple calls
title = AXUIElementCopyAttributeValue(element, 'AXTitle', None)
role = AXUIElementCopyAttributeValue(element, 'AXRole', None)

# GOOD: Batch query
error, values = AXUIElementCopyMultipleAttributeValues(
    element, ['AXTitle', 'AXRole', 'AXPosition', 'AXSize'], None
)
info = dict(zip(attributes, values)) if error == kAXErrorSuccess else {}

# BAD: Observer for every notification without debounce

# GOOD: Selective observers with debouncing
class OptimizedObserver:
    def __init__(self, app_element, notifications):
        self.last_callback, self.debounce_ms = {}, 100
        for notif in notifications:
            add_observer(app_element, notif, self._debounced_callback)

    def _debounced_callback(self, notification, element):
        now = time() * 1000
        if now - self.last_callback.get(notification, 0) < self.debounce_ms:
            return
        self.last_callback[notification] = now
        self._handle_notification(notification, element)