"""Safe Menu Navigation with Accessibility API Integration.
Provides comprehensive menu automation with security validation,
path-based navigation, and robust error handling.
"""
from __future__ import annotations
import asyncio
import re
import time
from typing import TYPE_CHECKING, Any
from ..core.contracts import require
from ..core.types import Duration
from ..integration.km_client import Either, KMError
# Module-level constants for default values to avoid B008
_DEFAULT_MENU_TIMEOUT = Duration.from_seconds(5)
_DEFAULT_STRUCTURE_TIMEOUT = Duration.from_seconds(10)
if TYPE_CHECKING:
from .app_controller import AppIdentifier
class MenuNavigator:
"""Safe menu navigation with accessibility API integration.
Provides comprehensive menu automation with:
- Path-based menu navigation with validation
- Security checks for menu item safety
- Timeout protection for hanging operations
- Error recovery and fallback strategies
"""
def __init__(self):
# Menu security patterns to block dangerous items
self._dangerous_menu_patterns = [
r"delete.*all",
r"format.*disk",
r"erase.*disk",
r"reset.*system",
r"factory.*reset",
r"sudo",
r"terminal",
r"shell",
# Add more dangerous patterns as needed
]
# Menu navigation cache for performance
self._menu_cache: dict[str, dict[str, Any]] = {}
self._cache_timeout = 30.0 # seconds
@require(lambda app_id: app_id.primary_identifier() != "")
@require(lambda path: len(path) > 0)
async def navigate_menu(
self,
app_id: AppIdentifier,
menu_path: list[str],
timeout: Duration | None = None,
) -> Either[KMError, bool]:
"""Navigate menu hierarchy with error recovery.
Security Features:
- Menu path validation and sanitization
- Dangerous menu item blocking
- Safe AppleScript generation with escaping
- Timeout protection for hanging operations
Architecture:
- Pattern: Strategy Pattern with multiple navigation methods
- Security: Input validation and injection prevention
- Performance: Menu structure caching with intelligent invalidation
"""
# Handle default timeout
if timeout is None:
timeout = _DEFAULT_MENU_TIMEOUT
try:
# Phase 1: Validate menu path
validation_result = self._validate_menu_path(menu_path)
if validation_result.is_left():
return validation_result
# Phase 2: Check for dangerous menu items
safety_check = self._check_menu_safety(menu_path)
if safety_check.is_left():
return safety_check
# Phase 3: Navigate menu via AppleScript
navigation_result = await self._navigate_applescript(
app_id,
menu_path,
timeout,
)
if navigation_result.is_left():
return navigation_result
return Either.right(True)
except Exception as e:
return Either.left(
KMError.execution_error(f"Menu navigation failed: {e!s}"),
)
# FIXME: Contract disabled - @require(lambda menu_path: len(menu_path) > 0)
def _validate_menu_path(self, menu_path: list[str]) -> Either[KMError, bool]:
"""Validate menu path for security and format.
Checks for:
- Valid string format and length
- Dangerous characters and patterns
- Path depth limits
"""
if len(menu_path) > 10: # Reasonable depth limit
return Either.left(
KMError.validation_error(
f"Menu path too deep (max 10): {len(menu_path)}",
),
)
for i, item in enumerate(menu_path):
# Basic validation
if not isinstance(item, str):
return Either.left(
KMError.validation_error(
f"Menu item {i} must be string: {type(item)}",
),
)
if len(item) == 0:
return Either.left(
KMError.validation_error(f"Menu item {i} cannot be empty"),
)
if len(item) > 100:
return Either.left(
KMError.validation_error(
f"Menu item {i} too long (max 100): {len(item)}",
),
)
# Security validation - check for dangerous characters
if self._contains_dangerous_chars(item):
return Either.left(
KMError.validation_error(
f"Menu item {i} contains dangerous characters: {item}",
),
)
return Either.right(True)
def _check_menu_safety(self, menu_path: list[str]) -> Either[KMError, bool]:
"""Check menu path for dangerous operations."""
full_path = " ".join(menu_path).lower()
for pattern in self._dangerous_menu_patterns:
if re.search(pattern, full_path, re.IGNORECASE):
return Either.left(
KMError.validation_error(
f"Dangerous menu operation blocked: {pattern}",
),
)
return Either.right(True)
def _contains_dangerous_chars(self, menu_item: str) -> bool:
"""Check for dangerous characters in menu item."""
dangerous_chars = [
'"',
"'",
"\\",
"\n",
"\r",
"\t", # AppleScript injection
";",
"|",
"&",
"$", # Shell injection
"<",
">",
"`", # Command injection
]
return any(char in menu_item for char in dangerous_chars)
async def _navigate_applescript(
self,
app_id: AppIdentifier,
menu_path: list[str],
timeout: Duration,
) -> Either[KMError, bool]:
"""Navigate menu via AppleScript UI automation.
Generates safe AppleScript with proper escaping and error handling.
"""
try:
# Escape menu items for AppleScript
escaped_items = [
self._escape_applescript_string(item) for item in menu_path
]
# Build nested menu navigation script
script = self._build_menu_script(app_id, escaped_items)
# Execute with timeout
result = await self._execute_menu_script(script, timeout)
return result
except Exception as e:
return Either.left(
KMError.execution_error(f"AppleScript navigation failed: {e!s}"),
)
def _escape_applescript_string(self, value: str) -> str:
"""Escape string for safe use in AppleScript."""
if not isinstance(value, str):
value = str(value)
# Replace dangerous characters
value = value.replace("\\", "\\\\") # Escape backslashes
value = value.replace('"', '\\"') # Escape quotes
value = value.replace("\n", "\\n") # Escape newlines
value = value.replace("\r", "\\r") # Escape carriage returns
value = value.replace("\t", "\\t") # Escape tabs
return value
def _build_menu_script(
self,
app_id: AppIdentifier,
escaped_items: list[str],
) -> str:
"""Build AppleScript for menu navigation."""
app_name = self._escape_applescript_string(app_id.display_name())
# Build nested menu access
menu_access = "menu bar 1"
for item in escaped_items:
menu_access = f'menu "{item}" of {menu_access}'
script = f"""
tell application "System Events"
try
tell application "{app_name}" to activate
delay 0.5
tell process "{app_name}"
-- Navigate to menu item
click {menu_access}
return "SUCCESS"
end tell
on error errorMessage
return "ERROR: " & errorMessage
end try
end tell
"""
return script
async def _execute_menu_script(
self,
script: str,
timeout: Duration,
) -> Either[KMError, bool]:
"""Execute menu navigation script with timeout."""
try:
# Execute AppleScript asynchronously
process = await asyncio.create_subprocess_exec(
"osascript",
"-e",
script,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
# Wait for completion with timeout
try:
stdout, stderr = await asyncio.wait_for(
process.communicate(),
timeout=timeout.total_seconds(),
)
except asyncio.TimeoutError:
# Kill the process if it times out
process.terminate()
await process.wait()
return Either.left(KMError.timeout_error(timeout))
# Check execution result
if process.returncode != 0:
error_msg = stderr.decode().strip() if stderr else "Unknown error"
return Either.left(
KMError.execution_error(f"Menu script failed: {error_msg}"),
)
result = stdout.decode().strip()
if result.startswith("ERROR:"):
return Either.left(KMError.execution_error(result[6:].strip()))
return Either.right(True)
except Exception as e:
return Either.left(
KMError.execution_error(f"Script execution failed: {e!s}"),
)
def _find_menu_item(self, _menu_name: str, _parent_menu: Any) -> Any | None:
"""Find menu item in menu structure - placeholder for accessibility API."""
# This would integrate with macOS accessibility APIs
# Implementation depends on specific accessibility framework used
return None
def _click_menu_item(self, _menu_item: Any) -> bool:
"""Click menu item with accessibility API - placeholder."""
# This would integrate with macOS accessibility APIs
# Implementation depends on specific accessibility framework used
return True
async def get_menu_structure(
self,
_app_id: AppIdentifier,
timeout: Duration | None = None,
) -> Either[KMError, dict[str, Any]]:
"""Get application menu structure for inspection.
Returns hierarchical menu structure for analysis and validation.
"""
# Handle default timeout
if timeout is None:
timeout = _DEFAULT_STRUCTURE_TIMEOUT
try:
# This would query the application's menu structure
# via accessibility APIs or AppleScript
# Placeholder implementation - would be expanded in production
structure = {
"menu_bar": [],
"context_menus": [],
"app_menu": [],
"timestamp": time.time(),
}
return Either.right(structure)
except Exception as e:
return Either.left(
KMError.execution_error(f"Menu structure query failed: {e!s}"),
)
def clear_menu_cache(self, app_id: AppIdentifier | None = None) -> bool:
"""Clear menu structure cache."""
if app_id:
cache_key = app_id.primary_identifier()
if cache_key in self._menu_cache:
del self._menu_cache[cache_key]
else:
self._menu_cache.clear()
def _is_menu_cache_valid(self, app_id: AppIdentifier) -> bool:
"""Check if menu cache is still valid."""
cache_key = app_id.primary_identifier()
if cache_key not in self._menu_cache:
return False
cache_data = self._menu_cache[cache_key]
cache_time = cache_data.get("timestamp", 0)
return time.time() - cache_time < self._cache_timeout
