oai/oai/mcp/platform.py

"""
Cross-platform MCP configuration for oAI.

This module handles OS-specific configuration, path handling,
and security checks for the MCP filesystem server.
"""

import os
import platform
import subprocess
from pathlib import Path
from typing import List, Dict, Any, Optional

from oai.constants import SYSTEM_DIRS_BLACKLIST
from oai.utils.logging import get_logger


class CrossPlatformMCPConfig:
    """
    Handle OS-specific MCP configuration.

    Provides methods for path normalization, security validation,
    and OS-specific default directories.

    Attributes:
        system: Operating system name
        is_macos: Whether running on macOS
        is_linux: Whether running on Linux
        is_windows: Whether running on Windows
    """

    def __init__(self):
        """Initialize platform detection."""
        self.system = platform.system()
        self.is_macos = self.system == "Darwin"
        self.is_linux = self.system == "Linux"
        self.is_windows = self.system == "Windows"

        logger = get_logger()
        logger.info(f"Detected OS: {self.system}")

    def get_default_allowed_dirs(self) -> List[Path]:
        """
        Get safe default directories for the current OS.

        Returns:
            List of default directories that are safe to access
        """
        home = Path.home()

        if self.is_macos:
            return [
                home / "Documents",
                home / "Desktop",
                home / "Downloads",
            ]

        elif self.is_linux:
            dirs = [home / "Documents"]

            # Try to get XDG directories
            try:
                for xdg_dir in ["DOCUMENTS", "DESKTOP", "DOWNLOAD"]:
                    result = subprocess.run(
                        ["xdg-user-dir", xdg_dir],
                        capture_output=True,
                        text=True,
                        timeout=1
                    )
                    if result.returncode == 0:
                        dir_path = Path(result.stdout.strip())
                        if dir_path.exists():
                            dirs.append(dir_path)
            except (subprocess.TimeoutExpired, FileNotFoundError):
                # Fallback to standard locations
                dirs.extend([
                    home / "Desktop",
                    home / "Downloads",
                ])

            return list(set(dirs))

        elif self.is_windows:
            return [
                home / "Documents",
                home / "Desktop",
                home / "Downloads",
            ]

        # Fallback for unknown OS
        return [home]

    def get_python_command(self) -> str:
        """
        Get the Python executable path.

        Returns:
            Path to the Python executable
        """
        import sys
        return sys.executable

    def get_filesystem_warning(self) -> str:
        """
        Get OS-specific security warning message.

        Returns:
            Warning message for the current OS
        """
        if self.is_macos:
            return """
Note: macOS Security
The Filesystem MCP server needs access to your selected folder.
You may see a security prompt - click 'Allow' to proceed.
(System Settings > Privacy & Security > Files and Folders)
"""
        elif self.is_linux:
            return """
Note: Linux Security
The Filesystem MCP server will access your selected folder.
Ensure oAI has appropriate file permissions.
"""
        elif self.is_windows:
            return """
Note: Windows Security
The Filesystem MCP server will access your selected folder.
You may need to grant file access permissions.
"""
        return ""

    def normalize_path(self, path: str) -> Path:
        """
        Normalize a path for the current OS.

        Expands user directory (~) and resolves to absolute path.

        Args:
            path: Path string to normalize

        Returns:
            Normalized absolute Path
        """
        return Path(os.path.expanduser(path)).resolve()

    def is_system_directory(self, path: Path) -> bool:
        """
        Check if a path is a protected system directory.

        Args:
            path: Path to check

        Returns:
            True if the path is a system directory
        """
        path_str = str(path)
        for blocked in SYSTEM_DIRS_BLACKLIST:
            if path_str.startswith(blocked):
                return True
        return False

    def is_safe_path(self, requested_path: Path, allowed_dirs: List[Path]) -> bool:
        """
        Check if a path is within allowed directories.

        Args:
            requested_path: Path being requested
            allowed_dirs: List of allowed parent directories

        Returns:
            True if the path is within an allowed directory
        """
        try:
            requested = requested_path.resolve()

            for allowed in allowed_dirs:
                try:
                    allowed_resolved = allowed.resolve()
                    requested.relative_to(allowed_resolved)
                    return True
                except ValueError:
                    continue

            return False
        except Exception:
            return False

    def get_folder_stats(self, folder: Path) -> Dict[str, Any]:
        """
        Get statistics for a folder.

        Args:
            folder: Path to the folder

        Returns:
            Dictionary with folder statistics:
            - exists: Whether the folder exists
            - file_count: Number of files (if exists)
            - total_size: Total size in bytes (if exists)
            - size_mb: Size in megabytes (if exists)
            - error: Error message (if any)
        """
        logger = get_logger()

        try:
            if not folder.exists() or not folder.is_dir():
                return {"exists": False}

            file_count = 0
            total_size = 0

            for item in folder.rglob("*"):
                if item.is_file():
                    file_count += 1
                    try:
                        total_size += item.stat().st_size
                    except (OSError, PermissionError):
                        pass

            return {
                "exists": True,
                "file_count": file_count,
                "total_size": total_size,
                "size_mb": total_size / (1024 * 1024),
            }

        except Exception as e:
            logger.error(f"Error getting folder stats for {folder}: {e}")
            return {"exists": False, "error": str(e)}