#!/usr/bin/env python3 """Install public thin plugins or a versioned local runtime. No credentials required in open mode.""" import argparse import hashlib import json import os import re import shutil import stat import subprocess import sys import tempfile import urllib.parse import urllib.request import zipfile from pathlib import Path, PurePosixPath ORIGIN = "https://skill.lionbaker.com" MARKETPLACE = "lionbaker-skills" PLUGINS = {"ai-alchemist-director", "auto-edit-talking-video", "short-video-traffic", "taiwan-health-compliance"} MAX_DOWNLOAD = 30 * 1024 * 1024 MAX_EXTRACTED = 80 * 1024 * 1024 MARKER = ".lionbaker-install.json" def sha256(data): return hashlib.sha256(data).hexdigest() def download(url, limit=MAX_DOWNLOAD): parsed = urllib.parse.urlsplit(url) if parsed.scheme != "https" or parsed.netloc != "skill.lionbaker.com" or parsed.username: raise ValueError("僅接受 skill.lionbaker.com 的 HTTPS 下載") req = urllib.request.Request(url, headers={"User-Agent": "LionBaker-Skills-Installer/0.1", "Cache-Control": "no-cache"}) with urllib.request.urlopen(req, timeout=90) as response: final = urllib.parse.urlsplit(response.url) if final.scheme != "https" or final.netloc != parsed.netloc: raise ValueError("下載跳轉到未允許的來源") data = response.read(limit + 1) if len(data) > limit: raise ValueError("下載超過預期大小") return data def extract(archive, destination, runtime=False): destination.mkdir(parents=True, exist_ok=False) total, seen, checksums = 0, set(), {} with zipfile.ZipFile(archive) as package: if len(package.infolist()) > 10000: raise ValueError("安裝包項目過多") for entry in package.infolist(): parts = entry.filename.split("/") rel = PurePosixPath(entry.filename) mode = entry.external_attr >> 16 if (rel.is_absolute() or "\\" in entry.filename or any(p in ("", ".", "..") or ":" in p for p in parts) or entry.filename in seen or stat.S_ISLNK(mode) or entry.is_dir()): raise ValueError("安裝包包含不安全或重複路徑") if any(p in ("node_modules", "private", ".git", ".env") for p in parts): raise ValueError("安裝包含禁止發佈的資料") if runtime and ("references" in parts or rel.name == "SKILL.md"): raise ValueError("Runtime 不得包含教材鏡像") total += entry.file_size if total > MAX_EXTRACTED: raise ValueError("解壓後超過預期大小") target = destination.joinpath(*parts) if not target.resolve().is_relative_to(destination.resolve()): raise ValueError("解壓路徑超出目錄") data = package.read(entry) target.parent.mkdir(parents=True, exist_ok=True) target.write_bytes(data) checksums[entry.filename] = sha256(data) seen.add(entry.filename) return checksums def run(codex, *args): done = subprocess.run([codex, *args], text=True, capture_output=True) if done.returncode: raise RuntimeError(f"Codex 指令失敗（{done.returncode}）：{done.stderr.strip() or done.stdout.strip()}") return json.loads(done.stdout) def replace_managed(stage, destination, metadata): if destination.is_symlink(): raise ValueError("目標目錄不可為符號連結") if destination.exists(): marker = destination / MARKER if not marker.is_file() or json.loads(marker.read_text()).get("marketplace") != MARKETPLACE: raise ValueError("目標已有非本安裝器管理的檔案，停止以避免覆蓋") (stage / MARKER).write_text(json.dumps({"marketplace": MARKETPLACE, **metadata}, ensure_ascii=False, indent=2)) backup = destination.with_name(destination.name + ".previous") if backup.exists(): raise ValueError("存在上次更新備份，請先檢查後再更新") destination.parent.mkdir(parents=True, exist_ok=True) moved = destination.exists() if moved: destination.rename(backup) try: stage.rename(destination) except Exception: if moved: backup.rename(destination) raise if moved: shutil.rmtree(backup) def install(args, manifest): expected = manifest["package"]["sha256"] if not re.fullmatch(r"[a-f0-9]{64}", expected) or manifest["marketplace"] != MARKETPLACE: raise ValueError("安裝清單格式不正確") names = [p["name"] for p in manifest["plugins"]] if len(names) != 4 or set(names) != PLUGINS: raise ValueError("安裝清單的外掛集合不符") codex = args.codex or shutil.which("codex") if not codex: raise RuntimeError("找不到 Codex CLI，請由 Codex 協助定位正式 CLI，再以 --codex 指定路徑") destination = (args.root or Path.home() / ".agents/marketplaces/lionbaker-skills").expanduser().absolute() configured = run(str(codex), "plugin", "marketplace", "list", "--json") matches = [m for m in configured.get("marketplaces", []) if m.get("name") == MARKETPLACE] if any(Path(m["root"]).resolve() != destination.resolve() for m in matches): raise ValueError("已註冊同名 marketplace 但路徑不同；停止並請 Codex 確認遷移，保留原設定") data = download(manifest["package"]["url"]) if sha256(data) != expected or len(data) != manifest["package"]["bytes"]: raise ValueError("安裝包 SHA-256 或大小不符，停止安裝") destination.parent.mkdir(parents=True, exist_ok=True) with tempfile.TemporaryDirectory(prefix=".lionbaker-install-", dir=destination.parent) as temp: archive = Path(temp) / "package.zip" archive.write_bytes(data) stage = Path(temp) / "stage" checksums = extract(archive, stage) catalog = json.loads((stage / ".agents/plugins/marketplace.json").read_text()) if catalog["name"] != MARKETPLACE or {p["name"] for p in catalog["plugins"]} != PLUGINS: raise ValueError("安裝包 marketplace 不符") replace_managed(stage, destination, {"release": manifest["release"], "sha256": expected, "files": checksums}) if not matches: run(str(codex), "plugin", "marketplace", "add", str(destination), "--json") installed = [] for plugin in manifest["plugins"]: run(str(codex), "plugin", "add", f"{plugin['name']}@{MARKETPLACE}", "--json") installed.append({"name": plugin["name"], "version": plugin["version"], "status": "installed"}) return {"marketplace": MARKETPLACE, "root": str(destination), "release": manifest["release"], "plugins": installed, "next": "重新啟動宿主或開新對話，確認載入 lionbaker-skills 來源。如另有同名獨立 Skill／personal 舊版，請確認停用舊入口以免觸發衝突；未改動其他來源或外掛。"} def install_runtime(args, manifest): if manifest.get("access_mode") != "open": raise ValueError("目前不允許匿名準備執行資源；請先透過外掛重新取得使用資格") if args.runtime not in PLUGINS or not re.fullmatch(r"[a-f0-9]{24}", args.release or "") or not re.fullmatch(r"[a-f0-9]{64}", args.sha256 or ""): raise ValueError("Runtime 需要登錄的外掛、有效 release 及 SHA-256") if args.release != manifest["release"]: raise ValueError("主機版本已更新；請重新呼叫 get_authorized_skill_content 與 get_runtime_package") base = args.root or Path(os.environ.get("LOCALAPPDATA", Path.home() / ".cache")) / "lionbaker-skills/runtime" destination = base.expanduser().absolute() / args.release / args.runtime marker = destination / MARKER if marker.is_file() and not destination.is_symlink(): previous = json.loads(marker.read_text()) if previous.get("sha256") == args.sha256 and previous.get("files") and all( (destination / p).is_file() and not (destination / p).is_symlink() and sha256((destination / p).read_bytes()) == expected for p, expected in previous["files"].items() ): return {"runtime_root": str(destination), "release": args.release, "cached": True} data = download(f"{ORIGIN}/runtime/{args.release}/{args.runtime}.zip") if sha256(data) != args.sha256: raise ValueError("Runtime SHA-256 不符，停止執行") destination.parent.mkdir(parents=True, exist_ok=True) with tempfile.TemporaryDirectory(prefix=".lionbaker-runtime-", dir=destination.parent) as temp: archive = Path(temp) / "runtime.zip" archive.write_bytes(data) stage = Path(temp) / "stage" checksums = extract(archive, stage, runtime=True) replace_managed(stage, destination, {"release": args.release, "sha256": args.sha256, "files": checksums}) return {"runtime_root": str(destination), "release": args.release, "cached": False, "next": "以此實際路徑作為 runtime_root；完整流程與 references 仍由 MCP 提供。Node.js、FFmpeg、字型及選用 Whisper 依目前技能流程準備。"} def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--root", type=Path, help="使用自訂的持久安裝／runtime 根目錄") parser.add_argument("--codex", help="正式 Codex CLI 路徑；預設從 PATH 定位") parser.add_argument("--runtime", choices=sorted(PLUGINS)) parser.add_argument("--release") parser.add_argument("--sha256") args = parser.parse_args() try: manifest = json.loads(download(f"{ORIGIN}/install/manifest.json", 1024 * 1024)) result = install_runtime(args, manifest) if args.runtime else install(args, manifest) print(json.dumps(result, ensure_ascii=False, indent=2)) except Exception as exc: print(f"安裝停止：{exc}", file=sys.stderr) return 1 return 0 if __name__ == "__main__": sys.exit(main())