diff --git a/patcher/analyze.py b/patcher/analyze.py index ac5699a..ba273ab 100644 --- a/patcher/analyze.py +++ b/patcher/analyze.py @@ -4,13 +4,15 @@ """ import pathlib import platform +import re import subprocess +import time -from . import defio, discovery, llm as llm_mod, ui +from . import bytecode, defio, discovery, llm as llm_mod, ui -def probe_version(binary: str): - """跑 binary --version 拿版本号;失败(如被 Gatekeeper 拦)返回 None。""" +def _try_probe(binary: str): + """单次 binary --version 尝试,成功返回版本串,失败返回 None。""" try: r = subprocess.run( [binary, "--version"], capture_output=True, text=True, timeout=30 @@ -21,6 +23,60 @@ def probe_version(binary: str): return None +def probe_version(binary: str): + """跑 binary --version 拿版本号;首次失败等 2 秒重试一次(macOS Gatekeeper + 首次执行新路径的二进制可能需要额外的公证检查时间)。""" + v = _try_probe(binary) + if v: + return v + time.sleep(2) + return _try_probe(binary) + + +def remove_quarantine(binary: str): + """去掉 macOS quarantine 属性,避免 Gatekeeper 拦截。""" + if platform.system() == "Darwin": + subprocess.run(["xattr", "-d", "com.apple.quarantine", binary], + capture_output=True) + + +_SEMVER_QUOTED_RE = re.compile(rb'"(\d{1,2}\.\d{1,2}\.\d{1,4})"') +_SEMVER_RAW_RE = re.compile(rb'(\d{1,2}\.\d{1,2}\.\d{1,4})') + + +def probe_version_from_data(data: bytes): + """从二进制数据中提取版本号(当 --version 探测失败时的降级方案)。 + + 先扫描源码文本中带引号的 semver 模式;bytecode 构建上再验证常量池 + 唯一性,找不到时降级扫不带引号的模式(常量池条目是原始字节)。 + """ + candidates = set() + for m in _SEMVER_QUOTED_RE.finditer(data): + v = m.group(1).decode("ascii", errors="ignore") + if int(v.split(".")[0]) <= 19: + candidates.add(v) + is_bc = bytecode.is_bytecode_build(data) + if is_bc: + pool_hits = [v for v in candidates if bytecode.unique(data, v)] + if len(pool_hits) == 1: + return pool_hits[0] + extra = set() + for m in _SEMVER_RAW_RE.finditer(data): + v = m.group(1).decode("ascii", errors="ignore") + if v not in candidates and int(v.split(".")[0]) <= 19: + extra.add(v) + pool_hits = [v for v in extra if bytecode.unique(data, v)] + if len(pool_hits) == 1: + return pool_hits[0] + return None + if not candidates: + return None + shortest = min(candidates, key=len) + if len(shortest) <= 10: + return shortest + return None + + def safe_label(label: str) -> str: """版本标签用于文件名: 过滤 shell/glob 危险字符(如 probe 失败的 '?')。""" cleaned = "".join(ch if ch.isalnum() or ch in "._-" else "_" for ch in label) @@ -88,9 +144,7 @@ def run(binary: str, cfg: dict, project_root: pathlib.Path, ui.info(f"{binary} " + ui.dim(f"({len(data):,} bytes)")) # cp/下载来的文件可能带 quarantine,先去掉,否则后面 --version 探测会被 Gatekeeper 杀 - if platform.system() == "Darwin": - subprocess.run(["xattr", "-d", "com.apple.quarantine", binary], - capture_output=True) + remove_quarantine(binary) display = cfg.get("display", {}) raw_name = display.get("name", "") @@ -132,6 +186,10 @@ def run(binary: str, cfg: dict, project_root: pathlib.Path, ui.step("发现锚点") probed = probe_version(binary) + if not probed: + probed = probe_version_from_data(data) + if probed: + ui.info(f"--version 探测失败,从二进制数据提取到版本: {probed}") patches, report, problems, warnings = discovery.build_patches( data, name, version_arg, design, theme_color=theme_color, real_version=probed, diff --git a/patcher/cli.py b/patcher/cli.py index 91d1484..a7f9706 100644 --- a/patcher/cli.py +++ b/patcher/cli.py @@ -66,6 +66,15 @@ def _cmd_all(args, cfg): from . import pipeline binary = _resolve_binary(args) + if args.bundle: + if not args.apply: + from . import ui + ui.fail("--bundle 必须搭配 --apply 使用") + return 1 + return pipeline.bundle(binary, cfg, PROJECT_ROOT, + output_dir=args.output_dir, suffix=args.suffix, + label=args.label, out=args.out, use_llm=args.llm, + force=args.force, yes=args.yes, pty=args.pty) if not args.apply: rc = pipeline.run(binary, cfg, PROJECT_ROOT, label=args.label, out=args.out, use_llm=args.llm, @@ -84,7 +93,9 @@ def _cmd_skin(args, cfg): from . import interactive return interactive.run(getattr(args, "binary", None), cfg, PROJECT_ROOT, - yes=args.yes, pty=args.pty) + yes=args.yes, pty=args.pty, + bundle=args.bundle, suffix=args.suffix, + output_dir=args.output_dir) def _cmd_preview(args, cfg): @@ -122,6 +133,10 @@ def main(argv=None): p.add_argument("binary", nargs="?", help="不填则自动定位系统安装") p.add_argument("-y", "--yes", action="store_true", help="跳过最终确认") p.add_argument("--pty", action="store_true", help="最后抓真实启动画面验证") + p.add_argument("--bundle", action="store_true", + help="创建版本目录,放入原始 + 换肤二进制") + p.add_argument("--suffix", default="skin", help="换肤文件后缀(默认 skin)") + p.add_argument("--output-dir", help="bundle 输出目录(默认在源文件旁建 v版本号/)") p.set_defaults(func=_cmd_skin) p = sub.add_parser("analyze", help="发现锚点,生成补丁定义 TOML") @@ -158,6 +173,10 @@ def main(argv=None): help="覆盖已存在的定义文件(默认写 .draft.toml)") p.add_argument("--pty", action="store_true") p.add_argument("-y", "--yes", action="store_true", help="跳过写入前的交互确认") + p.add_argument("--bundle", action="store_true", + help="创建版本目录,放入原始 + 换肤二进制") + p.add_argument("--suffix", default="skin", help="换肤文件后缀(默认 skin)") + p.add_argument("--output-dir", help="bundle 输出目录(默认在源文件旁建 v版本号/)") p.set_defaults(func=_cmd_all) p = sub.add_parser("preview", help="渲染图标设计稿预览") diff --git a/patcher/interactive.py b/patcher/interactive.py index 77631b7..a41daf9 100644 --- a/patcher/interactive.py +++ b/patcher/interactive.py @@ -100,7 +100,8 @@ def _ask_hex() -> str: ui.warn(str(e)) -def run(binary, cfg, project_root: pathlib.Path, yes=False, pty=False) -> int: +def run(binary, cfg, project_root: pathlib.Path, yes=False, pty=False, + bundle=False, suffix="skin", output_dir=None) -> int: if not sys.stdin.isatty(): ui.fail("skin 是交互向导,请在终端里运行;非交互请用 all --apply -y") return 1 @@ -173,8 +174,13 @@ def run(binary, cfg, project_root: pathlib.Path, yes=False, pty=False) -> int: "real_version": not version_s, } - rc = pipeline.run(path, new_cfg, project_root, - label=version, yes=True, pty=pty) + if bundle: + rc = pipeline.bundle(path, new_cfg, project_root, + output_dir=output_dir, suffix=suffix, + label=version, yes=True, pty=pty) + else: + rc = pipeline.run(path, new_cfg, project_root, + label=version, yes=True, pty=pty) if rc == 0: _save_config(project_root, new_cfg) return rc diff --git a/patcher/pipeline.py b/patcher/pipeline.py index fb00cb3..2117660 100644 --- a/patcher/pipeline.py +++ b/patcher/pipeline.py @@ -3,6 +3,7 @@ cli 的 `all` 命令和交互向导 `skin` 都走这里,避免两份拷贝各自漂移。 """ import pathlib +import shutil from . import analyze, patch as patch_mod, sign, ui, verify @@ -12,6 +13,7 @@ def run(binary: str, cfg: dict, project_root: pathlib.Path, *, force: bool = False, yes: bool = False, pty: bool = False, apply: bool = True) -> int: """全流程执行;apply=False 时只做 analyze(dry-run)。返回退出码。""" + analyze.remove_quarantine(binary) label = analyze.safe_label(label or analyze.probe_version(binary) or "unknown") out = analyze.resolve_out(project_root, label, out=out, force=force) # out 已定好(含 .draft 防覆盖),force=True 让 analyze 直接写它 @@ -30,3 +32,50 @@ def run(binary: str, cfg: dict, project_root: pathlib.Path, *, ui.step("全部完成") ui.ok(f"可以运行了: {binary}") return rc + + +def bundle(binary: str, cfg: dict, project_root: pathlib.Path, *, + output_dir: str = None, suffix: str = "skin", **kw) -> int: + """创建版本目录,放入原始 + 换肤二进制。 + + 目录结构: v{版本号}/{原始文件名} + v{版本号}/{原始文件名}{suffix} + """ + analyze.remove_quarantine(binary) + src = pathlib.Path(binary) + ver = analyze.probe_version(binary) + if not ver: + data = src.read_bytes() + ver = analyze.probe_version_from_data(data) + if not ver: + ui.fail("版本探测失败,无法确定目录名;请用 --label 显式指定") + return 1 + + bundle_dir = pathlib.Path(output_dir) if output_dir else src.parent / f"v{ver}" + if bundle_dir.exists(): + ui.warn(f"目录已存在: {bundle_dir}") + bundle_dir.mkdir(parents=True, exist_ok=True) + + orig_path = bundle_dir / src.name + skin_path = bundle_dir / (src.stem + suffix + src.suffix) + if not src.suffix: + skin_path = bundle_dir / (src.name + suffix) + + ui.step("创建 bundle") + shutil.copy2(binary, orig_path) + ui.ok(f"原始文件: {orig_path}") + shutil.copy2(binary, skin_path) + ui.info(f"待换肤: {skin_path}") + + rc = run(str(skin_path), cfg, project_root, **kw) + if rc != 0: + return rc + + backup = pathlib.Path(str(skin_path) + ".orig") + if backup.exists(): + backup.unlink() + + ui.step("bundle 就绪") + ui.ok(f"目录: {bundle_dir}") + ui.info(f" 原始: {orig_path.name}") + ui.info(f" 换肤: {skin_path.name}") + return 0 diff --git a/tests/test_core.py b/tests/test_core.py index 8607033..1cbc75d 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -7,14 +7,16 @@ import os import struct import pathlib +import shutil import sys import tempfile import unittest +from unittest import mock sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent)) -from patcher import (bytecode, defio, discovery, glyphpack, locate, - patch as patch_mod, tomlmini) +from patcher import (analyze, bytecode, defio, discovery, glyphpack, locate, + patch as patch_mod, pipeline, tomlmini) def esc(s: str) -> str: @@ -634,5 +636,131 @@ def test_sniff(self): self.assertEqual(locate.sniff_kind(str(pathlib.Path(td) / "x")), "unknown") +class TestProbeVersionRetry(unittest.TestCase): + def test_succeeds_on_second_attempt(self): + calls = [] + + def fake_probe(binary): + calls.append(1) + return None if len(calls) == 1 else "2.1.281" + + with mock.patch("patcher.analyze._try_probe", side_effect=fake_probe), \ + mock.patch("patcher.analyze.time.sleep"): + result = analyze.probe_version("/fake/binary") + self.assertEqual(result, "2.1.281") + self.assertEqual(len(calls), 2) + + def test_returns_none_after_both_fail(self): + with mock.patch("patcher.analyze._try_probe", return_value=None), \ + mock.patch("patcher.analyze.time.sleep"): + result = analyze.probe_version("/fake/binary") + self.assertIsNone(result) + + def test_returns_immediately_on_first_success(self): + with mock.patch("patcher.analyze._try_probe", return_value="1.0.0"): + result = analyze.probe_version("/fake/binary") + self.assertEqual(result, "1.0.0") + + +class TestProbeVersionFromData(unittest.TestCase): + def test_bytecode_build(self): + data = make_bytecode_bundle(version="2.1.281") + result = analyze.probe_version_from_data(data) + self.assertEqual(result, "2.1.281") + + def test_non_bytecode_build(self): + data = b'"3.0.42"' + os.urandom(200) + result = analyze.probe_version_from_data(data) + self.assertEqual(result, "3.0.42") + + def test_no_match(self): + data = os.urandom(500) + result = analyze.probe_version_from_data(data) + self.assertIsNone(result) + + def test_rejects_high_major_version(self): + data = b'"99.1.0"' + os.urandom(200) + result = analyze.probe_version_from_data(data) + self.assertIsNone(result) + + +class TestBundle(unittest.TestCase): + def _make_patchable_binary(self, td): + """创建一个可供 pipeline 打补丁的合成二进制(只需能通过 analyze)。""" + binary = pathlib.Path(td) / "claude" + data = make_bytecode_bundle(version="2.1.300") + binary.write_bytes(data) + binary.chmod(0o755) + return str(binary) + + def test_bundle_creates_directory_structure(self): + with tempfile.TemporaryDirectory() as td: + binary = self._make_patchable_binary(td) + project = pathlib.Path(__file__).resolve().parent.parent + bundle_dir = pathlib.Path(td) / "v2.1.300" + + with mock.patch("patcher.analyze.probe_version", return_value="2.1.300"), \ + mock.patch("patcher.pipeline.run", return_value=0): + rc = pipeline.bundle(binary, {}, project) + + self.assertEqual(rc, 0) + self.assertTrue(bundle_dir.exists()) + self.assertTrue((bundle_dir / "claude").exists()) + self.assertTrue((bundle_dir / "claudeskin").exists()) + + def test_bundle_custom_suffix(self): + with tempfile.TemporaryDirectory() as td: + binary = self._make_patchable_binary(td) + project = pathlib.Path(__file__).resolve().parent.parent + + with mock.patch("patcher.analyze.probe_version", return_value="2.1.300"), \ + mock.patch("patcher.pipeline.run", return_value=0): + rc = pipeline.bundle(binary, {}, project, suffix="mod") + + bundle_dir = pathlib.Path(td) / "v2.1.300" + self.assertEqual(rc, 0) + self.assertTrue((bundle_dir / "claudemod").exists()) + + def test_bundle_custom_output_dir(self): + with tempfile.TemporaryDirectory() as td: + binary = self._make_patchable_binary(td) + project = pathlib.Path(__file__).resolve().parent.parent + custom = pathlib.Path(td) / "custom_out" + + with mock.patch("patcher.analyze.probe_version", return_value="2.1.300"), \ + mock.patch("patcher.pipeline.run", return_value=0): + rc = pipeline.bundle(binary, {}, project, output_dir=str(custom)) + + self.assertEqual(rc, 0) + self.assertTrue(custom.exists()) + self.assertTrue((custom / "claude").exists()) + self.assertTrue((custom / "claudeskin").exists()) + + def test_bundle_fails_on_probe_failure(self): + with tempfile.TemporaryDirectory() as td: + binary = pathlib.Path(td) / "claude" + binary.write_bytes(os.urandom(500)) + project = pathlib.Path(__file__).resolve().parent.parent + + with mock.patch("patcher.analyze.probe_version", return_value=None), \ + mock.patch("patcher.analyze.probe_version_from_data", return_value=None): + rc = pipeline.bundle(str(binary), {}, project) + + self.assertEqual(rc, 1) + + def test_bundle_original_untouched(self): + with tempfile.TemporaryDirectory() as td: + binary = self._make_patchable_binary(td) + original_data = pathlib.Path(binary).read_bytes() + project = pathlib.Path(__file__).resolve().parent.parent + bundle_dir = pathlib.Path(td) / "v2.1.300" + + with mock.patch("patcher.analyze.probe_version", return_value="2.1.300"), \ + mock.patch("patcher.pipeline.run", return_value=0): + pipeline.bundle(binary, {}, project) + + self.assertEqual((bundle_dir / "claude").read_bytes(), original_data) + + if __name__ == "__main__": unittest.main()