#!/usr/bin/env python3 """Synchronize allowlisted skills from their latest Gitea Release assets.""" from __future__ import annotations import argparse import json import os import re import shutil import stat import sys import tempfile import urllib.error import urllib.parse import urllib.request import zipfile from pathlib import Path, PurePosixPath from typing import Any REPO_ROOT = Path(__file__).resolve().parents[1] DEFAULT_SOURCES = REPO_ROOT / ".dist-sync" / "sources.json" DEFAULT_VERSIONS = REPO_ROOT / ".dist-sync" / "versions.json" MAX_ASSET_BYTES = 250 * 1024 * 1024 MAX_API_BYTES = 5 * 1024 * 1024 MAX_ARCHIVE_FILES = 10_000 MAX_ARCHIVE_DEPTH = 32 MAX_ARCHIVE_PATH_LENGTH = 1024 TAG_PATTERN = re.compile(r"v[0-9]+(?:\.[0-9]+)*") class SyncError(RuntimeError): """A source exists but cannot be synchronized safely.""" def log(message: str) -> None: print(message, flush=True) def read_json(path: Path) -> Any: try: return json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as error: raise SyncError(f"无法读取 JSON：{path}: {error}") from error def write_json_atomic(path: Path, value: Any) -> None: path.parent.mkdir(parents=True, exist_ok=True) encoded = (json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True) + "\n").encode("utf-8") temporary: Path | None = None try: with tempfile.NamedTemporaryFile(prefix=f".{path.name}.", dir=path.parent, delete=False) as output: temporary = Path(output.name) output.write(encoded) output.flush() os.fsync(output.fileno()) os.replace(temporary, path) except OSError as error: if temporary is not None: temporary.unlink(missing_ok=True) raise SyncError(f"无法原子更新 JSON：{path}: {error}") from error def validate_source(raw: Any) -> dict[str, str]: if not isinstance(raw, dict): raise SyncError("sources.json 的每个来源必须是对象") source: dict[str, str] = {} for field in ("name", "repository", "asset"): value = raw.get(field) if not isinstance(value, str) or not value.strip(): raise SyncError(f"来源缺少有效字段：{field}") source[field] = value.strip() name = source["name"] repository = source["repository"] asset = source["asset"] if not re.fullmatch(r"[a-z0-9]+(?:-[a-z0-9]+)*", name): raise SyncError(f"无效 skill 名称：{name}") owner, repo = repository.split("/", 1) if "/" in repository else ("", "") repository_part = re.compile(r"[A-Za-z0-9_.-]+") if not owner or not repo or not repository_part.fullmatch(owner) or not repository_part.fullmatch(repo): raise SyncError(f"无效仓库名称：{repository}") if Path(asset).name != asset: raise SyncError(f"附件名称不能包含路径：{asset}") return source def canonical_origin(url: str) -> tuple[str, str, int | None]: parsed = urllib.parse.urlsplit(url) scheme = parsed.scheme.lower() hostname = parsed.hostname.lower() if parsed.hostname else "" try: port = parsed.port except ValueError as error: raise SyncError(f"URL 端口无效：{url}") from error if scheme not in {"http", "https"} or not hostname: raise SyncError(f"URL 必须是有效的 HTTP(S) 地址：{url}") effective_port = port if port is not None else (443 if scheme == "https" else 80) return scheme, hostname, effective_port class SameOriginRedirectHandler(urllib.request.HTTPRedirectHandler): def __init__(self, allowed_origin: tuple[str, str, int | None]) -> None: super().__init__() self.allowed_origin = allowed_origin def redirect_request( self, request: urllib.request.Request, file_pointer: Any, code: int, message: str, headers: Any, new_url: str, ) -> urllib.request.Request | None: if canonical_origin(new_url) != self.allowed_origin: raise SyncError(f"拒绝跨源重定向：{request.full_url} -> {new_url}") return super().redirect_request(request, file_pointer, code, message, headers, new_url) def build_opener(server_url: str) -> tuple[urllib.request.OpenerDirector, tuple[str, str, int | None]]: origin = canonical_origin(server_url) return urllib.request.build_opener(SameOriginRedirectHandler(origin)), origin def request_json(url: str, token: str, opener: urllib.request.OpenerDirector) -> tuple[int, Any | None]: request = urllib.request.Request( url, headers={ "Accept": "application/json", "Authorization": f"token {token}", "User-Agent": "agent-skills-dist-sync", }, ) try: with opener.open(request, timeout=30) as response: body = response.read(MAX_API_BYTES + 1) if len(body) > MAX_API_BYTES: raise SyncError(f"Gitea API 响应超过 {MAX_API_BYTES} bytes：{url}") try: return response.status, json.loads(body) except (UnicodeDecodeError, json.JSONDecodeError) as error: raise SyncError(f"Gitea API 返回无效 JSON：{url}: {error}") from error except urllib.error.HTTPError as error: if error.code == 404: return error.code, None detail = error.read(2048).decode("utf-8", errors="replace").strip() raise SyncError(f"Gitea API 返回 HTTP {error.code}: {detail or url}") from error except SyncError: raise except OSError as error: raise SyncError(f"请求 Gitea API 失败：{url}: {error}") from error def download( url: str, token: str, destination: Path, opener: urllib.request.OpenerDirector, allowed_origin: tuple[str, str, int | None], ) -> None: if canonical_origin(url) != allowed_origin: raise SyncError(f"附件下载地址与 Gitea 不同源，拒绝发送 token：{url}") request = urllib.request.Request( url, headers={ "Accept": "application/octet-stream", "Authorization": f"token {token}", "User-Agent": "agent-skills-dist-sync", }, ) try: with opener.open(request, timeout=120) as response, destination.open("wb") as output: declared_size = response.headers.get("Content-Length") if declared_size: try: if int(declared_size) > MAX_ASSET_BYTES: raise SyncError(f"附件超过 {MAX_ASSET_BYTES} bytes：{url}") except ValueError as error: raise SyncError(f"附件返回无效 Content-Length：{declared_size!r}") from error copied = 0 while chunk := response.read(1024 * 1024): copied += len(chunk) if copied > MAX_ASSET_BYTES: raise SyncError(f"附件超过 {MAX_ASSET_BYTES} bytes：{url}") output.write(chunk) except urllib.error.HTTPError as error: detail = error.read(2048).decode("utf-8", errors="replace").strip() raise SyncError(f"下载附件返回 HTTP {error.code}: {detail or url}") from error except SyncError: raise except OSError as error: raise SyncError(f"下载附件失败：{url}: {error}") from error def safe_extract(archive_path: Path, destination: Path) -> None: try: with zipfile.ZipFile(archive_path) as archive: files = 0 total_size = 0 for info in archive.infolist(): normalized = info.filename.replace("\\", "/") member = PurePosixPath(normalized) if member.is_absolute() or ".." in member.parts: raise SyncError(f"ZIP 包含不安全路径：{info.filename}") if not member.parts: continue if len(normalized) > MAX_ARCHIVE_PATH_LENGTH or len(member.parts) > MAX_ARCHIVE_DEPTH: raise SyncError(f"ZIP 路径过长或目录过深：{info.filename}") mode = info.external_attr >> 16 if stat.S_ISLNK(mode): raise SyncError(f"ZIP 不允许符号链接：{info.filename}") target = destination.joinpath(*member.parts) if not info.is_dir(): files += 1 if files > MAX_ARCHIVE_FILES: raise SyncError(f"ZIP 文件数量超过 {MAX_ARCHIVE_FILES}") total_size += info.file_size if total_size > MAX_ASSET_BYTES: raise SyncError("ZIP 解压后内容过大") target.parent.mkdir(parents=True, exist_ok=True) with archive.open(info) as source, target.open("wb") as output: shutil.copyfileobj(source, output) if mode & 0o111: target.chmod(target.stat().st_mode | 0o111) else: target.mkdir(parents=True, exist_ok=True) if files == 0: raise SyncError("ZIP 中没有文件") except zipfile.BadZipFile as error: raise SyncError(f"附件不是有效 ZIP：{archive_path}") from error def frontmatter_name(skill_file: Path) -> str: try: lines = skill_file.read_text(encoding="utf-8-sig").splitlines() except (OSError, UnicodeDecodeError) as error: raise SyncError(f"无法读取 {skill_file}: {error}") from error if not lines or lines[0].strip() != "---": raise SyncError(f"SKILL.md 缺少 YAML frontmatter：{skill_file}") for line in lines[1:]: if line.strip() == "---": break if line.startswith("name:"): return line.split(":", 1)[1].strip().strip("'\"") raise SyncError(f"SKILL.md frontmatter 缺少 name：{skill_file}") def stage_source( source: dict[str, str], *, server_url: str, token: str, current_versions: dict[str, str], next_versions: dict[str, str], temp_root: Path, opener: urllib.request.OpenerDirector, allowed_origin: tuple[str, str, int | None], ) -> tuple[str, Path | None]: name = source["name"] repository = source["repository"] encoded_repository = "/".join(urllib.parse.quote(part, safe="") for part in repository.split("/")) latest_url = f"{server_url}/api/v1/repos/{encoded_repository}/releases/latest" status, release = request_json(latest_url, token, opener) if status == 404: log(f"跳过 {name}：仓库不可见或尚无 latest Release") return "skipped", None if not isinstance(release, dict): raise SyncError(f"{repository} 的 latest Release 响应无效") tag = release.get("tag_name") assets = release.get("assets") if not isinstance(tag, str) or not TAG_PATTERN.fullmatch(tag): raise SyncError(f"{repository} 的 latest Release tag 不符合 v* 正式版本约定：{tag!r}") if not isinstance(assets, list): raise SyncError(f"{repository} 的 latest Release 缺少附件列表") matches = [asset for asset in assets if isinstance(asset, dict) and asset.get("name") == source["asset"]] if len(matches) != 1: raise SyncError(f"{repository}@{tag} 应恰好包含一个附件 {source['asset']}，实际为 {len(matches)} 个") target = REPO_ROOT / "skills" / name if current_versions.get(name) == tag and target.is_dir(): log(f"跳过 {name}：已经同步 {tag}") return "unchanged", None download_url = matches[0].get("browser_download_url") if not isinstance(download_url, str) or not download_url: raise SyncError(f"{repository}@{tag} 的附件缺少 browser_download_url") work_dir = temp_root / name extract_dir = work_dir / "content" archive_path = work_dir / source["asset"] extract_dir.mkdir(parents=True) download(download_url, token, archive_path, opener, allowed_origin) safe_extract(archive_path, extract_dir) skill_file = extract_dir / "SKILL.md" if not skill_file.is_file(): raise SyncError(f"{repository}@{tag} 的 ZIP 根目录缺少 SKILL.md") actual_name = frontmatter_name(skill_file) if actual_name != name: raise SyncError(f"{repository}@{tag} 的 skill name 为 {actual_name!r}，预期 {name!r}") next_versions[name] = tag log(f"已准备 {name}：{current_versions.get(name, '未记录')} -> {tag}") return "updated", extract_dir def apply_staged_updates( staged: list[tuple[str, Path]], current_versions: dict[str, str], next_versions: dict[str, str], versions_path: Path, ) -> None: if not staged: return backup_root = Path(tempfile.mkdtemp(prefix=".dist-sync-backup-", dir=REPO_ROOT)) applied: list[tuple[Path, Path | None]] = [] try: for name, staged_dir in staged: destination = REPO_ROOT / "skills" / name replacement = destination.with_name(f".{name}.dist-sync-new") if replacement.exists(): shutil.rmtree(replacement) shutil.copytree(staged_dir, replacement) backup: Path | None = None if destination.exists(): backup = backup_root / name destination.rename(backup) try: replacement.rename(destination) except Exception: if backup is not None and backup.exists() and not destination.exists(): backup.rename(destination) raise applied.append((destination, backup)) write_json_atomic(versions_path, next_versions) except Exception as error: for destination, backup in reversed(applied): if destination.exists(): shutil.rmtree(destination) if backup is not None and backup.exists(): backup.rename(destination) try: write_json_atomic(versions_path, current_versions) except SyncError as rollback_error: raise SyncError( f"应用同步结果失败；skill 目录已回滚，但 versions.json 回滚失败：{rollback_error}" ) from error if isinstance(error, SyncError): raise raise SyncError(f"应用同步结果失败，已回滚 skill 目录：{error}") from error finally: shutil.rmtree(backup_root, ignore_errors=True) for name, _ in staged: replacement = REPO_ROOT / "skills" / f".{name}.dist-sync-new" if replacement.exists(): shutil.rmtree(replacement, ignore_errors=True) def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--sources", type=Path, default=DEFAULT_SOURCES) parser.add_argument("--versions", type=Path, default=DEFAULT_VERSIONS) parser.add_argument("--server-url", default=os.environ.get("GITEA_SERVER_URL", "")) parser.add_argument("--token", default=os.environ.get("SOURCE_REPOS_TOKEN", "")) args = parser.parse_args() server_url = args.server_url.rstrip("/") token = args.token.strip() if not server_url: raise SyncError("缺少 GITEA_SERVER_URL 或 --server-url") if not token: raise SyncError("缺少 SOURCE_REPOS_TOKEN 或 --token") opener, allowed_origin = build_opener(server_url) config = read_json(args.sources) raw_sources = config.get("sources") if isinstance(config, dict) else None if not isinstance(raw_sources, list) or not raw_sources: raise SyncError("sources.json 必须包含非空 sources 数组") sources = [validate_source(raw) for raw in raw_sources] names = [source["name"] for source in sources] if len(names) != len(set(names)): raise SyncError("sources.json 包含重复的 skill name") raw_versions = read_json(args.versions) if args.versions.exists() else {} if not isinstance(raw_versions, dict) or not all( isinstance(name, str) and isinstance(tag, str) and TAG_PATTERN.fullmatch(tag) for name, tag in raw_versions.items() ): raise SyncError("versions.json 必须是 skill name 到正式 v* Release tag 的字符串映射") current_versions: dict[str, str] = dict(raw_versions) next_versions = dict(current_versions) counts = {"updated": 0, "unchanged": 0, "skipped": 0} staged: list[tuple[str, Path]] = [] with tempfile.TemporaryDirectory(prefix="agent-skills-dist-sync-") as temp: for source in sources: result, staged_dir = stage_source( source, server_url=server_url, token=token, current_versions=current_versions, next_versions=next_versions, temp_root=Path(temp), opener=opener, allowed_origin=allowed_origin, ) counts[result] += 1 if staged_dir is not None: staged.append((source["name"], staged_dir)) apply_staged_updates(staged, current_versions, next_versions, args.versions) log( "同步完成：" f"更新 {counts['updated']}，无变化 {counts['unchanged']}，" f"无 latest Release {counts['skipped']}" ) return 0 if __name__ == "__main__": try: raise SystemExit(main()) except SyncError as error: print(f"同步失败：{error}", file=sys.stderr) raise SystemExit(1)