MCP Weather & Accruals Server

#!/usr/bin/env python3 """ Модуль для работы с файлами. Содержит функции для чтения, создания и изменения файлов по определённому пути. """ import logging import os from pathlib import Path from typing import Optional logger = logging.getLogger("FileAPI") # Разрешённая базовая директория для операций с файлами. # Значение теперь читается из переменной окружения `ALLOWED_BASE_DIR`. # По умолчанию сохранено прежнее поведение для обратной совместимости. ALLOWED_BASE_DIR = os.getenv( "ALLOWED_BASE_DIR", "/Users/aleksandrhohon/Downloads/Скриншоты" ) # Проверяет, является ли путь безопасным (не выходит за пределы base_dir). def is_path_safe(file_path: str, base_dir: str = ALLOWED_BASE_DIR) -> bool: """ Проверяет, является ли путь безопасным (не выходит за пределы base_dir). Предотвращает path traversal атаки. Args: file_path: Проверяемый путь base_dir: Базовая директория (по умолчанию текущая директория) Returns: bool: True если путь безопасен, False иначе """ try: base = Path(base_dir).resolve() target = (base / file_path).resolve() # Проверяем, что целевой путь находится в базовой директории return str(target).startswith(str(base)) except Exception as e: logger.error(f"Ошибка при проверке пути: {e}") return False # Читает содержимое файла. async def read_file(file_path: str, base_dir: str = ALLOWED_BASE_DIR) -> str: """ Читает содержимое файла. Args: file_path: Путь к файлу (относительно base_dir) base_dir: Базовая директория для безопасности Returns: Содержимое файла или сообщение об ошибке """ logger.info(f"Чтение файла: {file_path}") # Проверяем безопасность пути if not is_path_safe(file_path, base_dir): error_msg = f"❌ Доступ запрещен: путь '{file_path}' выходит за пределы разрешённой директории" logger.warning(error_msg) return error_msg try: full_path = Path(base_dir) / file_path # Проверяем, существует ли файл if not full_path.exists(): error_msg = f"❌ Файл не найден: {file_path}" logger.warning(error_msg) return error_msg # Проверяем, что это файл (не директория) if not full_path.is_file(): error_msg = f"❌ Путь '{file_path}' не является файлом" logger.warning(error_msg) return error_msg # Читаем файл with open(full_path, 'r', encoding='utf-8') as f: content = f.read() logger.info(f"✅ Файл успешно прочитан: {file_path} ({len(content)} символов)") return content except Exception as e: error_msg = f"❌ Ошибка при чтении файла: {str(e)}" logger.error(error_msg) return error_msg # Создаёт новый файл с указанным содержимым. async def create_file(file_path: str, content: str, base_dir: str = ALLOWED_BASE_DIR, overwrite: bool = False) -> str: """ Создаёт новый файл с указанным содержимым. Args: file_path: Путь к файлу (относительно base_dir) content: Содержимое файла base_dir: Базовая директория для безопасности overwrite: Перезаписать ли файл, если он существует Returns: Сообщение об успехе или ошибке """ logger.info(f"Создание файла: {file_path}") # Проверяем безопасность пути if not is_path_safe(file_path, base_dir): error_msg = f"❌ Доступ запрещен: путь '{file_path}' выходит за пределы разрешённой директории" logger.warning(error_msg) return error_msg try: full_path = Path(base_dir) / file_path # Проверяем, существует ли файл if full_path.exists() and not overwrite: error_msg = f"❌ Файл уже существует: {file_path}. Используйте overwrite=True для перезаписи" logger.warning(error_msg) return error_msg # Создаём директории, если их нет full_path.parent.mkdir(parents=True, exist_ok=True) # Записываем файл with open(full_path, 'w', encoding='utf-8') as f: f.write(content) success_msg = f"✅ Файл успешно создан: {file_path} ({len(content)} символов)" logger.info(success_msg) return success_msg except Exception as e: error_msg = f"❌ Ошибка при создании файла: {str(e)}" logger.error(error_msg) return error_msg # Обновляет содержимое существующего файла. async def update_file(file_path: str, content: str, base_dir: str = ALLOWED_BASE_DIR) -> str: """ Обновляет содержимое существующего файла. Args: file_path: Путь к файлу (относительно base_dir) content: Новое содержимое файла base_dir: Базовая директория для безопасности Returns: Сообщение об успехе или ошибке """ logger.info(f"Обновление файла: {file_path}") # Проверяем безопасность пути if not is_path_safe(file_path, base_dir): error_msg = f"❌ Доступ запрещен: путь '{file_path}' выходит за пределы разрешённой директории" logger.warning(error_msg) return error_msg try: full_path = Path(base_dir) / file_path # Проверяем, существует ли файл if not full_path.exists(): error_msg = f"❌ Файл не найден: {file_path}. Используйте create_file для создания нового файла" logger.warning(error_msg) return error_msg # Проверяем, что это файл (не директория) if not full_path.is_file(): error_msg = f"❌ Путь '{file_path}' не является файлом" logger.warning(error_msg) return error_msg # Читаем старое содержимое (для логирования) with open(full_path, 'r', encoding='utf-8') as f: old_size = len(f.read()) # Записываем новое содержимое with open(full_path, 'w', encoding='utf-8') as f: f.write(content) success_msg = f"✅ Файл успешно обновлен: {file_path} ({old_size} → {len(content)} символов)" logger.info(success_msg) return success_msg except Exception as e: error_msg = f"❌ Ошибка при обновлении файла: {str(e)}" logger.error(error_msg) return error_msg # Добавляет содержимое в конец существующего файла. async def append_to_file(file_path: str, content: str, base_dir: str = ALLOWED_BASE_DIR) -> str: """ Добавляет содержимое в конец существующего файла. Args: file_path: Путь к файлу (относительно base_dir) content: Содержимое для добавления base_dir: Базовая директория для безопасности Returns: Сообщение об успехе или ошибке """ logger.info(f"Добавление содержимого в файл: {file_path}") # Проверяем безопасность пути if not is_path_safe(file_path, base_dir): error_msg = f"❌ Доступ запрещен: путь '{file_path}' выходит за пределы разрешённой директории" logger.warning(error_msg) return error_msg try: full_path = Path(base_dir) / file_path # Проверяем, существует ли файл if not full_path.exists(): error_msg = f"❌ Файл не найден: {file_path}" logger.warning(error_msg) return error_msg # Добавляем содержимое в конец файла with open(full_path, 'a', encoding='utf-8') as f: f.write(content) success_msg = f"✅ Содержимое добавлено в файл: {file_path} ({len(content)} символов)" logger.info(success_msg) return success_msg except Exception as e: error_msg = f"❌ Ошибка при добавлении в файл: {str(e)}" logger.error(error_msg) return error_msg # Удаляет файл. async def delete_file(file_path: str, base_dir: str = ALLOWED_BASE_DIR) -> str: """ Удаляет файл. Args: file_path: Путь к файлу (относительно base_dir) base_dir: Базовая директория для безопасности Returns: Сообщение об успехе или ошибке """ logger.info(f"Удаление файла: {file_path}") # Проверяем безопасность пути if not is_path_safe(file_path, base_dir): error_msg = f"❌ Доступ запрещен: путь '{file_path}' выходит за пределы разрешённой директории" logger.warning(error_msg) return error_msg try: full_path = Path(base_dir) / file_path # Проверяем, существует ли файл if not full_path.exists(): error_msg = f"❌ Файл не найден: {file_path}" logger.warning(error_msg) return error_msg # Проверяем, что это файл (не директория) if not full_path.is_file(): error_msg = f"❌ Путь '{file_path}' не является файлом" logger.warning(error_msg) return error_msg # Удаляем файл full_path.unlink() success_msg = f"✅ Файл успешно удалён: {file_path}" logger.info(success_msg) return success_msg except Exception as e: error_msg = f"❌ Ошибка при удалении файла: {str(e)}" logger.error(error_msg) return error_msg # Выводит список файлов в директории. async def list_files(directory: str = ".", base_dir: str = ALLOWED_BASE_DIR) -> str: """ Выводит список файлов в директории. Args: directory: Директория для просмотра (относительно base_dir) base_dir: Базовая директория для безопасности Returns: Отформатированный список файлов или сообщение об ошибке """ logger.info(f"Получение списка файлов: {directory}") # Проверяем безопасность пути if not is_path_safe(directory, base_dir): error_msg = f"❌ Доступ запрещен: путь '{directory}' выходит за пределы разрешённой директории" logger.warning(error_msg) return error_msg try: full_path = Path(base_dir) / directory # Проверяем, существует ли директория if not full_path.exists(): error_msg = f"❌ Директория не найдена: {directory}" logger.warning(error_msg) return error_msg # Проверяем, что это директория (не файл) if not full_path.is_dir(): error_msg = f"❌ Путь '{directory}' не является директорией" logger.warning(error_msg) return error_msg # Получаем список файлов files = [] dirs = [] for item in sorted(full_path.iterdir()): if item.is_file(): size = item.stat().st_size files.append(f" 📄 {item.name} ({size} байт)") elif item.is_dir(): dirs.append(f" 📁 {item.name}/") result = f"📂 Содержимое директории: {directory}\n\n" if dirs: result += "Директории:\n" + "\n".join(dirs) + "\n\n" if files: result += "Файлы:\n" + "\n".join(files) if not dirs and not files: result += "(пусто)" logger.info(f"Успешно получен список файлов: {directory}") return result except Exception as e: error_msg = f"❌ Ошибка при получении списка файлов: {str(e)}" logger.error(error_msg) return error_msg