Добавлена загрузка файлов из Gitea в рабочую директорию

gitea_manager.py

@@ -1,8 +1,8 @@
 """
 title: Gitea Manager
 author: Claude
-version: 1.1.0
+version: 1.2.0
-description: Инструмент для управления Gitea из Open WebUI — создание и удаление репозиториев, организаций, файлов, веток и безопасная распаковка ZIP-архивов.
+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(
+        GITEA_URL: str = Field(default="http://192.168.31.100:30008", description="Базовый URL Gitea без слэша в конце.")
-            default="http://192.168.31.100:30008",
+        GITEA_TOKEN: str = Field(default="", description="API-токен Gitea. Заполните в настройках инструмента.")
-            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(
+        WORKING_DIRECTORY: str = Field(default="/tmp/gitea-manager-workdir", description="Корневая рабочая директория.")
-            default="/tmp/gitea-manager-workdir",
+        MAX_ARCHIVE_FILES: int = Field(default=10000, description="Максимальное количество элементов ZIP.")
-            description="Корневая рабочая директория для распаковки архивов.",
+        MAX_ARCHIVE_SIZE_MB: int = Field(default=2048, description="Максимальный распакованный размер ZIP в МБ.")
-        )
-        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()
@@ -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}",
+                method,
-                headers=headers, timeout=self.valves.TIMEOUT, **kwargs,
+                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 в рабочую директорию.
+        """Безопасно распаковать 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():
@@ -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():
+                        if not destination.is_dir() or destination.is_symlink():
-                            return f"Ошибка: путь назначения не является каталогом: {destination}"
+                            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()