268 lines
14 KiB
Python
268 lines
14 KiB
Python
"""
|
|
title: Gitea Manager
|
|
author: Claude
|
|
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
|
|
from pydantic import BaseModel, Field
|
|
|
|
|
|
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. Заполните в настройках инструмента.",
|
|
)
|
|
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()
|
|
|
|
def _check(self) -> Optional[str]:
|
|
"""Проверить доступность инструмента и наличие токена."""
|
|
if not self.valves.ENABLED:
|
|
return "Инструмент Gitea Manager выключен."
|
|
if not self.valves.GITEA_TOKEN:
|
|
return "API-токен Gitea не задан в настройках инструмента."
|
|
return None
|
|
|
|
def _request(self, method: str, path: str, **kwargs) -> str:
|
|
"""Выполнить запрос к Gitea и вернуть JSON или сообщение об ошибке."""
|
|
error = self._check()
|
|
if error:
|
|
return error
|
|
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,
|
|
)
|
|
if not response.ok:
|
|
return f"Ошибка {response.status_code}: {response.text}"
|
|
return response.text or "Операция выполнена успешно."
|
|
except requests.RequestException as exc:
|
|
return f"Ошибка соединения с Gitea: {exc}"
|
|
|
|
def create_repository(self, name: str, description: str = "", private: bool = False, auto_init: bool = True) -> str:
|
|
"""Создать репозиторий текущего пользователя."""
|
|
return self._request("POST", "/user/repos", json={"name": name, "description": description, "private": private, "auto_init": auto_init})
|
|
|
|
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})
|
|
|
|
def list_repositories(self) -> str:
|
|
"""Получить список репозиториев текущего пользователя."""
|
|
return self._request("GET", "/user/repos")
|
|
|
|
def get_repository_info(self, owner: str, repo: str) -> str:
|
|
"""Получить информацию о репозитории."""
|
|
return self._request("GET", f"/repos/{owner}/{repo}")
|
|
|
|
def delete_repository(self, owner: str, repo: str) -> str:
|
|
"""Удалить репозиторий. Операция необратима."""
|
|
return self._request("DELETE", f"/repos/{owner}/{repo}")
|
|
|
|
def create_organization(self, org_name: str, description: str = "", visibility: str = "public") -> str:
|
|
"""Создать организацию."""
|
|
return self._request("POST", "/orgs", json={"username": org_name, "description": description, "visibility": visibility})
|
|
|
|
def list_organizations(self) -> str:
|
|
"""Получить организации текущего пользователя."""
|
|
return self._request("GET", "/user/orgs")
|
|
|
|
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)
|
|
|
|
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)
|
|
try:
|
|
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):
|
|
return result
|
|
|
|
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
|
|
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
|
|
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
|
|
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))
|