197 lines
7.4 KiB
Python
197 lines
7.4 KiB
Python
#!/usr/bin/env python3
|
||
"""微信 APK/Agent/Hook 发布前置门禁。
|
||
|
||
只做本地资产校验：版本、APK 元数据、Agent 包、兼容矩阵和 arm64 Frida。
|
||
未在兼容矩阵中标记 verified 的微信版本，不得获得写类发布准入。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import hashlib
|
||
import json
|
||
import re
|
||
import shutil
|
||
import struct
|
||
import subprocess
|
||
import sys
|
||
import zipfile
|
||
from pathlib import Path
|
||
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
ANDROID_GRADLE = ROOT / "android-app" / "app" / "build.gradle"
|
||
AGENT_VERSION = ROOT / "agent" / "VERSION"
|
||
COMPAT = ROOT / "agent" / "hook" / "wechat_version_compat.json"
|
||
APK_DEFAULT = ROOT / "android-app" / "app" / "build" / "outputs" / "apk" / "release" / "app-release.apk"
|
||
FRIDA_ASSETS = (
|
||
ROOT / "scripts" / "frida-server-17.8.1-android-arm64",
|
||
ROOT / "scripts" / "frida-gadget-17.8.1-android-arm64.so",
|
||
)
|
||
|
||
|
||
def sha256(path: Path) -> str:
|
||
digest = hashlib.sha256()
|
||
with path.open("rb") as handle:
|
||
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
|
||
digest.update(chunk)
|
||
return digest.hexdigest()
|
||
|
||
|
||
def gradle_version() -> tuple[str, int]:
|
||
text = ANDROID_GRADLE.read_text(encoding="utf-8")
|
||
name = re.search(r'versionName\s+"([^"]+)"', text)
|
||
code = re.search(r"versionCode\s+(\d+)", text)
|
||
if not name or not code:
|
||
raise ValueError("Gradle 缺少 versionName/versionCode")
|
||
return name.group(1), int(code.group(1))
|
||
|
||
|
||
def apk_metadata(apk: Path) -> dict[str, str | int]:
|
||
tool = shutil.which("apkanalyzer")
|
||
if not tool:
|
||
raise RuntimeError("未找到 apkanalyzer，无法读取 APK 版本")
|
||
def read(field: str) -> str:
|
||
result = subprocess.run(
|
||
[tool, "manifest", field, str(apk)],
|
||
check=True,
|
||
capture_output=True,
|
||
text=True,
|
||
)
|
||
return result.stdout.strip()
|
||
return {
|
||
"application_id": read("application-id"),
|
||
"version_name": read("version-name"),
|
||
"version_code": int(read("version-code")),
|
||
}
|
||
|
||
|
||
def elf_machine(path: Path) -> int:
|
||
with path.open("rb") as handle:
|
||
header = handle.read(20)
|
||
if header[:4] != b"\x7fELF":
|
||
raise ValueError(f"不是 ELF 文件: {path}")
|
||
endian = "<" if header[5] == 1 else ">"
|
||
return struct.unpack_from(endian + "H", header, 18)[0]
|
||
|
||
|
||
def check_apk_assets(apk: Path) -> dict[str, object]:
|
||
required = ("assets/wechat_hook_v2.js", "assets/wechat_version_compat.json")
|
||
with zipfile.ZipFile(apk) as archive:
|
||
names = set(archive.namelist())
|
||
missing = [name for name in required if name not in names]
|
||
embedded = {}
|
||
for name in required:
|
||
if name in names:
|
||
embedded[name] = hashlib.sha256(archive.read(name)).hexdigest()
|
||
source_hashes = {
|
||
"assets/wechat_hook_v2.js": sha256(ROOT / "agent" / "hook" / "wechat_hook_v2.js"),
|
||
"assets/wechat_version_compat.json": sha256(COMPAT),
|
||
}
|
||
return {"missing": missing, "embedded_sha256": embedded, "source_sha256": source_hashes}
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser(description=__doc__)
|
||
parser.add_argument("--apk", type=Path, default=APK_DEFAULT)
|
||
parser.add_argument("--agent-archive", type=Path)
|
||
parser.add_argument("--wechat-version", help="待发布微信版本，例如 8.0.60")
|
||
parser.add_argument("--write", action="store_true", help="申请微信写类发布准入")
|
||
parser.add_argument("--json-out", type=Path)
|
||
args = parser.parse_args()
|
||
|
||
errors: list[str] = []
|
||
warnings: list[str] = []
|
||
apk = args.apk.resolve()
|
||
gradle_name, gradle_code = gradle_version()
|
||
agent_name = AGENT_VERSION.read_text(encoding="utf-8").strip()
|
||
metadata = apk_metadata(apk)
|
||
|
||
if metadata["version_name"] != gradle_name or metadata["version_code"] != gradle_code:
|
||
errors.append(
|
||
f"APK/Gradle 版本不一致: APK={metadata['version_name']}/{metadata['version_code']} "
|
||
f"Gradle={gradle_name}/{gradle_code}"
|
||
)
|
||
if agent_name != gradle_name:
|
||
errors.append(f"APK/Agent 版本不一致: APK={gradle_name} Agent={agent_name}")
|
||
|
||
asset_report = check_apk_assets(apk)
|
||
if asset_report["missing"]:
|
||
errors.append("APK 缺少微信 Hook 资产: " + ", ".join(asset_report["missing"]))
|
||
for key, embedded_hash in asset_report["embedded_sha256"].items():
|
||
if embedded_hash != asset_report["source_sha256"][key]:
|
||
errors.append(f"APK 内嵌资产与源文件不一致: {key}")
|
||
|
||
frida_report = {}
|
||
for asset in FRIDA_ASSETS:
|
||
if not asset.is_file():
|
||
errors.append(f"缺少 arm64 Frida asset: {asset}")
|
||
continue
|
||
machine = elf_machine(asset)
|
||
frida_report[str(asset.relative_to(ROOT.parent))] = {
|
||
"sha256": sha256(asset),
|
||
"elf_machine": machine,
|
||
"arm64": machine == 183,
|
||
}
|
||
if machine != 183:
|
||
errors.append(f"Frida asset 不是 AArch64: {asset}")
|
||
|
||
agent_archive_report: dict[str, object] = {}
|
||
if args.agent_archive:
|
||
with zipfile.ZipFile(args.agent_archive) if args.agent_archive.suffix == ".zip" else tarfile_open(args.agent_archive) as archive:
|
||
members = archive_names(archive)
|
||
agent_archive_report = {"path": str(args.agent_archive), "has_VERSION": "VERSION" in members}
|
||
if "VERSION" not in members:
|
||
errors.append("Agent 发布包缺少 VERSION")
|
||
|
||
compat = json.loads(COMPAT.read_text(encoding="utf-8"))
|
||
matrix = compat.get("versions", {})
|
||
write_gate = {"requested": args.write, "wechat_version": args.wechat_version, "allowed": False}
|
||
if args.write:
|
||
if not args.wechat_version:
|
||
errors.append("写类准入必须提供 --wechat-version")
|
||
else:
|
||
entry = matrix.get(args.wechat_version)
|
||
verified = bool(entry and entry.get("verified") is True)
|
||
write_gate["matrix_entry"] = entry
|
||
write_gate["allowed"] = verified
|
||
if not verified:
|
||
errors.append(f"微信 {args.wechat_version} 未通过兼容矩阵 verified，停止写类发布")
|
||
elif args.wechat_version and not matrix.get(args.wechat_version, {}).get("verified", False):
|
||
warnings.append(f"微信 {args.wechat_version} 仅可读类验证，写类仍锁定")
|
||
|
||
report = {
|
||
"status": "blocked" if errors else "pass",
|
||
"apk": {"path": str(apk), "sha256": sha256(apk), "metadata": metadata},
|
||
"gradle": {"version_name": gradle_name, "version_code": gradle_code},
|
||
"agent": {"version": agent_name},
|
||
"hook_assets": asset_report,
|
||
"frida_arm64_assets": frida_report,
|
||
"agent_archive": agent_archive_report,
|
||
"write_gate": write_gate,
|
||
"errors": errors,
|
||
"warnings": warnings,
|
||
}
|
||
print(json.dumps(report, ensure_ascii=False, indent=2))
|
||
if args.json_out:
|
||
args.json_out.parent.mkdir(parents=True, exist_ok=True)
|
||
args.json_out.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||
return 1 if errors else 0
|
||
|
||
|
||
def tarfile_open(path: Path):
|
||
import tarfile
|
||
return tarfile.open(path, "r:gz")
|
||
|
||
|
||
def archive_names(archive) -> set[str]:
|
||
names = set()
|
||
for member in archive.infolist() if hasattr(archive, "infolist") else archive.getmembers():
|
||
name = member.filename if hasattr(member, "filename") else member.name
|
||
names.add(name.lstrip("./"))
|
||
return names
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|