diff --git a/gitea_manager.py b/gitea_manager.py index 1d59fa2..cfd7752 100644 --- a/gitea_manager.py +++ b/gitea_manager.py @@ -1,8 +1,8 @@ """ title: Gitea Manager author: Claude -version: 1.1.0 -description: Инструмент для управления Gitea из Open WebUI — создание и удаление репозиториев, организаций, файлов, веток и безопасная распаковка ZIP-архивов. +version: 1.2.0 +description: Инструмент для управления Gitea из Open WebUI — создание и удаление репозиториев, организаций, файлов, веток, загрузка и безопасная распаковка ZIP-архивов. requirements: requests """ import base64 @@ -13,6 +13,7 @@ import tempfile import zipfile from pathlib import Path, PurePosixPath from typing import Optional +from urllib.parse import quote import requests from pydantic import BaseModel, Field @@ -22,28 +23,13 @@ class Tools: """Инструмент управления Gitea и локальными файлами рабочего пространства.""" class Valves(BaseModel): - GITEA_URL: str = Field( - default="http://192.168.31.100:30008", - description="Базовый URL Gitea без слэша в конце.", - ) - GITEA_TOKEN: str = Field( - default="", - description="API-токен Gitea. Заполните в настройках инструмента.", - ) + GITEA_URL: str = Field(default="http://192.168.31.100:30008", description="Базовый URL Gitea без слэша в конце.") + GITEA_TOKEN: str = Field(default="", description="API-токен Gitea. Заполните в настройках инструмента.") 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="Максимальный суммарный размер распакованных данных в мегабайтах.", - ) + 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="Максимальный распакованный размер ZIP в МБ.") def __init__(self): self.valves = self.Valves() @@ -64,8 +50,11 @@ class Tools: headers = {"Authorization": f"token {self.valves.GITEA_TOKEN}", "Content-Type": "application/json"} try: response = requests.request( - method, f"{self.valves.GITEA_URL.rstrip('/')}/api/v1{path}", - headers=headers, timeout=self.valves.TIMEOUT, **kwargs, + method, + f"{self.valves.GITEA_URL.rstrip('/')}/api/v1{path}", + headers=headers, + timeout=self.valves.TIMEOUT, + **kwargs, ) if not response.ok: return f"Ошибка {response.status_code}: {response.text}" @@ -79,7 +68,7 @@ class Tools: def create_repository_in_org(self, org: str, name: str, description: str = "", private: bool = False, auto_init: bool = True) -> str: """Создать репозиторий внутри организации.""" - return self._request("POST", f"/orgs/{org}/repos", json={"name": name, "description": description, "private": private, "auto_init": auto_init}) + return self._request("POST", f"/orgs/{quote(org, safe='')}/repos", json={"name": name, "description": description, "private": private, "auto_init": auto_init}) def list_repositories(self) -> str: """Получить список репозиториев текущего пользователя.""" @@ -87,11 +76,11 @@ class Tools: def get_repository_info(self, owner: str, repo: str) -> str: """Получить информацию о репозитории.""" - return self._request("GET", f"/repos/{owner}/{repo}") + return self._request("GET", f"/repos/{quote(owner, safe='')}/{quote(repo, safe='')}") def delete_repository(self, owner: str, repo: str) -> str: """Удалить репозиторий. Операция необратима.""" - return self._request("DELETE", f"/repos/{owner}/{repo}") + return self._request("DELETE", f"/repos/{quote(owner, safe='')}/{quote(repo, safe='')}") def create_organization(self, org_name: str, description: str = "", visibility: str = "public") -> str: """Создать организацию.""" @@ -104,12 +93,12 @@ class Tools: def list_files(self, owner: str, repo: str, path: str = "", branch: str = "") -> str: """Получить список файлов и папок.""" params = {"ref": branch} if branch else {} - return self._request("GET", f"/repos/{owner}/{repo}/contents/{path}", params=params) + return self._request("GET", f"/repos/{quote(owner, safe='')}/{quote(repo, safe='')}/contents/{quote(path, safe='/')}", params=params) def get_file_content(self, owner: str, repo: str, file_path: str, branch: str = "") -> str: """Получить содержимое файла и его SHA.""" params = {"ref": branch} if branch else {} - result = self._request("GET", f"/repos/{owner}/{repo}/contents/{file_path}", params=params) + result = self._request("GET", f"/repos/{quote(owner, safe='')}/{quote(repo, safe='')}/contents/{quote(file_path, safe='/')}", params=params) try: data = json.loads(result) content = base64.b64decode(data["content"]).decode("utf-8", errors="replace") @@ -122,42 +111,76 @@ class Tools: payload = {"content": base64.b64encode(content.encode()).decode(), "message": commit_message} if branch: payload["branch"] = branch - return self._request("POST", f"/repos/{owner}/{repo}/contents/{file_path}", json=payload) + return self._request("POST", f"/repos/{quote(owner, safe='')}/{quote(repo, safe='')}/contents/{quote(file_path, safe='/')}", 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 - return self._request("PUT", f"/repos/{owner}/{repo}/contents/{file_path}", json=payload) + return self._request("PUT", f"/repos/{quote(owner, safe='')}/{quote(repo, safe='')}/contents/{quote(file_path, safe='/')}", 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 - return self._request("DELETE", f"/repos/{owner}/{repo}/contents/{file_path}", json=payload) + return self._request("DELETE", f"/repos/{quote(owner, safe='')}/{quote(repo, safe='')}/contents/{quote(file_path, safe='/')}", json=payload) def list_branches(self, owner: str, repo: str) -> str: """Получить список веток репозитория.""" - return self._request("GET", f"/repos/{owner}/{repo}/branches") + return self._request("GET", f"/repos/{quote(owner, safe='')}/{quote(repo, safe='')}/branches") + + def download_repository_file_to_workspace(self, owner: str, repo: str, file_path: str, destination_name: str = "", branch: str = "", overwrite: bool = False) -> str: + """Скачать файл из Gitea в локальную рабочую директорию Open WebUI.""" + error = self._check() + if error: + return error + + try: + root = Path(self.valves.WORKING_DIRECTORY).expanduser().resolve() + root.mkdir(parents=True, exist_ok=True) + name = destination_name or Path(file_path).name + destination = (root / name).resolve() + if not self._is_inside(destination, root) or destination == root: + return "Ошибка безопасности: destination_name должен быть обычным именем внутри WORKING_DIRECTORY." + if destination.exists() and not overwrite: + return f"Ошибка: файл уже существует: {destination}. Укажи overwrite=true." + + params = {"ref": branch} if branch else {} + path = f"/repos/{quote(owner, safe='')}/{quote(repo, safe='')}/contents/{quote(file_path, safe='/')}" + headers = {"Authorization": f"token {self.valves.GITEA_TOKEN}"} + response = requests.get( + f"{self.valves.GITEA_URL.rstrip('/')}/api/v1{path}", + headers=headers, + params=params, + timeout=self.valves.TIMEOUT, + ) + if not response.ok: + return f"Ошибка {response.status_code}: {response.text}" + + data = response.json() + encoded = data.get("content", "").replace("\n", "") + raw = base64.b64decode(encoded, validate=True) + max_size = max(1, self.valves.MAX_ARCHIVE_SIZE_MB) * 1024 * 1024 + if len(raw) > max_size: + return f"Ошибка безопасности: размер файла превышает лимит {self.valves.MAX_ARCHIVE_SIZE_MB} МБ." + + temporary = destination.with_name(f".{destination.name}.tmp") + temporary.write_bytes(raw) + temporary.replace(destination) + return json.dumps({"status": "ok", "path": str(destination), "size": len(raw)}, ensure_ascii=False, indent=2) + except (OSError, ValueError, KeyError, requests.RequestException) as exc: + return f"Ошибка загрузки файла из Gitea: {exc}" def extract_zip(self, zip_path: str, destination_dir: str = "", overwrite: bool = False) -> str: - """Безопасно распаковать ZIP в рабочую директорию. - - Архив и путь назначения должны быть доступны внутри окружения Open WebUI. - По умолчанию файлы извлекаются в WORKING_DIRECTORY/<имя-архива>. - Защита блокирует Zip Slip, абсолютные пути, символические ссылки, - ZIP-бомбы и превышение заданных лимитов. - """ + """Безопасно распаковать 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(): @@ -165,7 +188,6 @@ class Tools: 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: @@ -174,23 +196,17 @@ class Tools: 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: + if relative.is_absolute() or ".." in relative.parts or (relative.parts and ":" in relative.parts[0]): 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} МБ." @@ -202,20 +218,20 @@ class Tools: 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}" + if not destination.is_dir() or destination.is_symlink(): + return f"Ошибка: путь назначения не является безопасным каталогом: {destination}" for item in temporary_dir.iterdir(): target = destination / item.name + if target.is_symlink(): + target.unlink() if target.exists() and target.is_dir() and item.is_dir(): self._merge_directory(item, target) else: @@ -227,19 +243,10 @@ class Tools: 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) + 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}" @@ -257,6 +264,8 @@ class Tools: destination.mkdir(parents=True, exist_ok=True) for item in source.iterdir(): target = destination / item.name + if target.is_symlink(): + target.unlink() if item.is_dir(): if target.exists() and not target.is_dir(): target.unlink()