From bc54ddad61ddefcedb57eb2744e6892d4cb23272 Mon Sep 17 00:00:00 2001 From: host Date: Mon, 3 Aug 2026 13:58:47 +0300 Subject: [PATCH] =?UTF-8?q?=D0=94=D0=BE=D0=B1=D0=B0=D0=B2=D0=BB=D0=B5?= =?UTF-8?q?=D0=BD=D0=B0=20=D0=B1=D0=B5=D0=B7=D0=BE=D0=BF=D0=B0=D1=81=D0=BD?= =?UTF-8?q?=D0=B0=D1=8F=20=D1=80=D0=B0=D1=81=D0=BF=D0=B0=D0=BA=D0=BE=D0=B2?= =?UTF-8?q?=D0=BA=D0=B0=20ZIP=20=D0=B2=20=D1=80=D0=B0=D0=B1=D0=BE=D1=87?= =?UTF-8?q?=D1=83=D1=8E=20=D0=B4=D0=B8=D1=80=D0=B5=D0=BA=D1=82=D0=BE=D1=80?= =?UTF-8?q?=D0=B8=D1=8E?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- gitea_manager.py | 159 ++++++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 152 insertions(+), 7 deletions(-) diff --git a/gitea_manager.py b/gitea_manager.py index 9c0fced..1d59fa2 100644 --- a/gitea_manager.py +++ b/gitea_manager.py @@ -1,11 +1,17 @@ """ title: Gitea Manager author: Claude -version: 1.0.0 -description: Инструмент для управления Gitea из Open WebUI — создание и удаление репозиториев, организаций, а также создание, чтение, обновление, удаление файлов и просмотр веток через Gitea API. +version: 1.1.0 +description: Инструмент для управления Gitea из Open WebUI — создание и удаление репозиториев, организаций, файлов, веток и безопасная распаковка ZIP-архивов. requirements: requests """ import base64 +import json +import shutil +import stat +import tempfile +import zipfile +from pathlib import Path, PurePosixPath from typing import Optional import requests @@ -13,7 +19,7 @@ from pydantic import BaseModel, Field class Tools: - """Инструмент управления Gitea.""" + """Инструмент управления Gitea и локальными файлами рабочего пространства.""" class Valves(BaseModel): GITEA_URL: str = Field( @@ -26,6 +32,18 @@ class Tools: ) TIMEOUT: int = Field(default=15, description="Таймаут запросов в секундах.") ENABLED: bool = Field(default=True, description="Включить инструмент Gitea Manager.") + WORKING_DIRECTORY: str = Field( + default="/tmp/gitea-manager-workdir", + description="Корневая рабочая директория для распаковки архивов.", + ) + MAX_ARCHIVE_FILES: int = Field( + default=10000, + description="Максимальное количество файлов и каталогов в ZIP-архиве.", + ) + MAX_ARCHIVE_SIZE_MB: int = Field( + default=2048, + description="Максимальный суммарный размер распакованных данных в мегабайтах.", + ) def __init__(self): self.valves = self.Valves() @@ -93,7 +111,7 @@ class Tools: params = {"ref": branch} if branch else {} result = self._request("GET", f"/repos/{owner}/{repo}/contents/{file_path}", params=params) try: - data = __import__("json").loads(result) + data = json.loads(result) content = base64.b64decode(data["content"]).decode("utf-8", errors="replace") return f"SHA: {data['sha']}\n\n{content}" except (ValueError, KeyError, TypeError): @@ -102,21 +120,148 @@ class Tools: def create_file(self, owner: str, repo: str, file_path: str, content: str, commit_message: str = "Добавлен файл", branch: str = "") -> str: """Создать файл в репозитории.""" payload = {"content": base64.b64encode(content.encode()).decode(), "message": commit_message} - if branch: payload["branch"] = branch + if branch: + payload["branch"] = branch return self._request("POST", f"/repos/{owner}/{repo}/contents/{file_path}", json=payload) def update_file(self, owner: str, repo: str, file_path: str, content: str, sha: str, commit_message: str = "Обновление файла", branch: str = "") -> str: """Обновить существующий файл по SHA.""" payload = {"content": base64.b64encode(content.encode()).decode(), "message": commit_message, "sha": sha} - if branch: payload["branch"] = branch + if branch: + payload["branch"] = branch return self._request("PUT", f"/repos/{owner}/{repo}/contents/{file_path}", json=payload) def delete_file(self, owner: str, repo: str, file_path: str, sha: str, commit_message: str = "Удаление файла", branch: str = "") -> str: """Удалить файл по SHA.""" payload = {"message": commit_message, "sha": sha} - if branch: payload["branch"] = branch + if branch: + payload["branch"] = branch return self._request("DELETE", f"/repos/{owner}/{repo}/contents/{file_path}", json=payload) def list_branches(self, owner: str, repo: str) -> str: """Получить список веток репозитория.""" return self._request("GET", f"/repos/{owner}/{repo}/branches") + + def extract_zip(self, zip_path: str, destination_dir: str = "", overwrite: bool = False) -> str: + """Безопасно распаковать ZIP в рабочую директорию. + + Архив и путь назначения должны быть доступны внутри окружения Open WebUI. + По умолчанию файлы извлекаются в WORKING_DIRECTORY/<имя-архива>. + Защита блокирует Zip Slip, абсолютные пути, символические ссылки, + ZIP-бомбы и превышение заданных лимитов. + """ + try: + archive = Path(zip_path).expanduser().resolve(strict=True) + if not archive.is_file(): + return "Ошибка: zip_path должен указывать на обычный файл." + + working_root = Path(self.valves.WORKING_DIRECTORY).expanduser().resolve() + working_root.mkdir(parents=True, exist_ok=True) + + if destination_dir: + destination = Path(destination_dir).expanduser() + if not destination.is_absolute(): + destination = working_root / destination + destination = destination.resolve() + else: + destination = working_root / archive.stem + + if not self._is_inside(destination, working_root): + return "Ошибка безопасности: destination_dir находится за пределами WORKING_DIRECTORY." + if destination.exists() and not overwrite: + return f"Ошибка: каталог назначения уже существует: {destination}. Укажи overwrite=true для слияния." + + max_files = max(1, self.valves.MAX_ARCHIVE_FILES) + max_size = max(1, self.valves.MAX_ARCHIVE_SIZE_MB) * 1024 * 1024 + total_size = 0 + + with zipfile.ZipFile(archive, "r") as zip_file: + members = zip_file.infolist() + if len(members) > max_files: + return f"Ошибка безопасности: архив содержит {len(members)} элементов, лимит — {max_files}." + + for info in members: + relative = PurePosixPath(info.filename) + if relative.is_absolute() or ".." in relative.parts: + return f"Ошибка безопасности: запрещённый путь в архиве: {info.filename}" + if relative.parts and ":" in relative.parts[0]: + return f"Ошибка безопасности: запрещённый путь в архиве: {info.filename}" + + mode = (info.external_attr >> 16) & 0xFFFF + if stat.S_ISLNK(mode): + return f"Ошибка безопасности: символическая ссылка в архиве запрещена: {info.filename}" + + total_size += info.file_size + if total_size > max_size: + return f"Ошибка безопасности: распакованный размер превышает лимит {self.valves.MAX_ARCHIVE_SIZE_MB} МБ." + + temporary_dir = Path(tempfile.mkdtemp(prefix=".zip-extract-", dir=working_root)) + try: + for info in members: + relative = PurePosixPath(info.filename) + target = (temporary_dir / Path(*relative.parts)).resolve() + if not self._is_inside(target, temporary_dir): + raise ValueError(f"Запрещённый путь в архиве: {info.filename}") + + if info.is_dir() or info.filename.endswith("/"): + target.mkdir(parents=True, exist_ok=True) + continue + + target.parent.mkdir(parents=True, exist_ok=True) + with zip_file.open(info, "r") as source, target.open("wb") as output: + shutil.copyfileobj(source, output, length=1024 * 1024) + + if destination.exists(): + if not destination.is_dir(): + return f"Ошибка: путь назначения не является каталогом: {destination}" + for item in temporary_dir.iterdir(): + target = destination / item.name + if target.exists() and target.is_dir() and item.is_dir(): + self._merge_directory(item, target) + else: + if target.exists() and target.is_dir(): + shutil.rmtree(target) + elif target.exists(): + target.unlink() + shutil.move(str(item), str(target)) + else: + temporary_dir.rename(destination) + temporary_dir = None + + return json.dumps({ + "status": "ok", + "message": "ZIP-архив успешно распакован.", + "archive": str(archive), + "destination": str(destination), + "items": len(members), + "uncompressed_size": total_size, + }, ensure_ascii=False, indent=2) + finally: + if temporary_dir is not None and temporary_dir.exists(): + shutil.rmtree(temporary_dir, ignore_errors=True) + + except (OSError, ValueError, zipfile.BadZipFile) as exc: + return f"Ошибка распаковки ZIP: {exc}" + + @staticmethod + def _is_inside(path: Path, root: Path) -> bool: + """Проверить, находится ли путь внутри корневого каталога.""" + try: + path.relative_to(root) + return True + except ValueError: + return False + + def _merge_directory(self, source: Path, destination: Path) -> None: + """Рекурсивно объединить временный каталог с существующим.""" + destination.mkdir(parents=True, exist_ok=True) + for item in source.iterdir(): + target = destination / item.name + if item.is_dir(): + if target.exists() and not target.is_dir(): + target.unlink() + self._merge_directory(item, target) + else: + if target.exists(): + target.unlink() + shutil.move(str(item), str(target))