441 lines
17 KiB
Python
441 lines
17 KiB
Python
#!/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)
|