From e2cb5e338edb687cdd1fec3374bd6aa479216c72 Mon Sep 17 00:00:00 2001 From: stefan Date: Fri, 28 Aug 2026 16:57:53 +0800 Subject: [PATCH 1/4] feat: add agent-friendly HAP deployment CLI --- .gitignore | 1 + CHANGELOG.md | 6 + PRIVACY.md | 9 +- README.md | 103 +++++-- SECURITY.md | 9 +- docs/PACKAGING.md | 7 +- hapsign/cli.py | 549 ++++++++++++++++++++++++++++++++--- hapsign/pipeline.py | 121 ++++++-- hapsign/signing/installer.py | 101 ++++++- sign_install.bat | 19 +- tests/test_cli.py | 321 +++++++++++++++++++- tests/test_config_paths.py | 12 + tests/test_pipeline.py | 162 +++++++++++ tests/test_signing.py | 111 +++++++ 14 files changed, 1413 insertions(+), 118 deletions(-) diff --git a/.gitignore b/.gitignore index a7a729c..f2dffcc 100644 --- a/.gitignore +++ b/.gitignore @@ -16,6 +16,7 @@ htmlcov/ # 签名产物（证书、密钥、Profile、签名后的 hap） signing_files/ +.hapsign/ signed_haps/ logs/ hapsign-config.json diff --git a/CHANGELOG.md b/CHANGELOG.md index 2a251ad..916c829 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,12 @@ ### Added +- 新增面向 Agent 的 `auth`、`devices list`、`sign`、`install`、`deploy` CLI + 子命令；支持单行 JSON stdout、stderr 日志、明确退出码与输入校验。 +- CLI 支持显式 HDC `--serial`、真机/模拟器候选标记、签名与安装分离，以及安装后 + `bm dump` 校验；Token 可跨目标设备复用，Profile 缓存按 UDID 隔离。 +- CLI 默认状态目录改为跨平台用户主目录 `~/.hapsign`，Windows 对应 + `%USERPROFILE%\.hapsign`，不再受 Agent 当前工作目录影响。 - macOS 支持：按平台解析 DevEco JBR / hap-sign-tool / hdc 路径，可用 `hapsign` 命令行签名安装。 - 可安装的 `hapsign` 命令和标准 Python 项目元数据。 - Ruff、pytest、覆盖率、pre-commit 和 Windows CI 配置。 diff --git a/PRIVACY.md b/PRIVACY.md index f11aa97..581c275 100644 --- a/PRIVACY.md +++ b/PRIVACY.md @@ -26,10 +26,13 @@ Profile。重复 UDID 会复用已有设备记录。用户应确认自己有权 ## 本地保存 -根据设置，以下文件保存在程序目录、用户 Local AppData 或用户选择的目录： +根据入口和设置，以下文件保存在 CLI 用户主目录、程序目录、用户 Local AppData +或用户选择的目录： -- `signing_files/.token_cache.json`：访问令牌、刷新令牌、JWT 和账号基本字段； -- `signing_files//`：私钥密钥库、CSR、证书、Profile 和缓存元数据； +- CLI 的 `~/.hapsign/.token_cache.json`（桌面/便携版为 + `signing_files/.token_cache.json`）：访问令牌、刷新令牌、JWT 和账号基本字段； +- CLI 的 `~/.hapsign//`（桌面/便携版为 `signing_files//`）： + 私钥密钥库、CSR、证书、Profile 和缓存元数据； - `signed_haps/`：可选保留的最后一个已签名 HAP； - `logs/hapsign.log*`：诊断日志； - `hapsign-config.json`：日志级别、保存位置和功能开关。 diff --git a/README.md b/README.md index 2be8c44..bfe25d9 100644 --- a/README.md +++ b/README.md @@ -131,6 +131,9 @@ set HAPSIGN_PYTHON=C:\path\to\your\python.exe $env:HAPSIGN_PYTHON = "C:\path\to\your\python.exe" ``` +拖拽脚本还需要明确的 HDC 目标序列号，可设 `HAPSIGN_SERIAL`，也可把序列号作为 +第二个参数传入。运行 `hapsign devices list` 可以查看候选设备。 + ## 使用 ### 方式一：桌面应用（推荐） @@ -164,18 +167,39 @@ hapsign-app ### 方式二：bat 拖拽 -将 `.hap` 文件直接拖到 `sign_install.bat` 上，自动完成签名+安装。 +先设置 `HAPSIGN_SERIAL`，再将 `.hap` 文件拖到 `sign_install.bat` 上；或从 CMD +显式传入 HAP 和设备序列号： + +```bat +set HAPSIGN_SERIAL=5XQ0225613000233 +sign_install.bat path\to\app-unsigned.hap 5XQ0225613000233 +``` ### 方式三：命令行（Windows / macOS） ```bash -hapsign --hap path/to/app-unsigned.hap -hapsign --hap path/to/app-signed.hap # 已签名则跳过签名，直接安装 +hapsign devices list --connected-only --json +hapsign auth status --json +hapsign auth --json +hapsign sign --hap path/to/app-unsigned.hap --serial --json +hapsign install --hap path/to/app-signed.hap --serial --json +hapsign deploy --hap path/to/app-unsigned.hap --serial --json ``` -包名会自动从 hap 内的 `module.json` 提取，无需手动指定。 -若 HAP 已包含签名块（Hap Signing Block），会跳过登录与签名，直接安装原文件。 -源码目录中仍可使用 `python main.py --hap ...`。 +`sign` 只签名并返回签名 HAP 的绝对路径；`install` 只接受已有 Hap Signing Block +的 HAP；`deploy` 端到端签名并安装，输入已经签名时会直接安装。包名默认从 HAP +里的 `module.json` 提取。源码目录中可用 `python3 main.py ...`。 + +所有执行命令都支持 `--json`。此模式下 stdout 只输出单行 JSON，日志写到 stderr， +且不会输出 Token、密码或 UDID。Agent 应先从 `devices list` 中选择 +`connected=true` 的目标，优先选择 `physical_candidate=true` 的 USB 真机，再把其 +`serial` 原样传给后续命令。`serial` 是 HDC 连接标识，不是签名 Profile 中的 UDID。 + +`auth` 可以单独调用并持久化当天 Token。同一份 Token 缓存不绑定目标设备，在同一 +台运行 HapSign 的电脑上可继续给不同 HarmonyOS 手机、平板或 PC 目标签名；每台 +目标设备的 Profile 仍绑定自己的 UDID，切换设备会重新申请 Profile。Token 不会在 +多台运行 HapSign 的电脑之间自动同步，也不建议手工复制缓存。`auth status` 只检查 +本地当日缓存，因此 JSON 中 `online_verified` 固定为 `false`。 ### 构建便携版 @@ -215,31 +239,48 @@ python scripts/build_portable.py --keep-bundled-browser > [开源发布门禁](docs/OPEN_SOURCE_RELEASE.md) 完成真实设备安装回归。 > `--allow-deveco-toolchain` 只用于排障回退，其产物不得冒充锁定的公开构建。 -### 完整参数 +### Agent CLI 接口 ``` -hapsign --hap [选项] - -选项: - --hap hap 文件路径（必填；已签名则直接安装） - --bundle-name 应用包名（不传则从 hap 内自动提取） - --country 国家码，默认 CN - --device-type 设备类型码，默认 4 - --work-dir 签名文件存储目录，默认 signing_files/{bundle_name}/ - --enable-capability 使用 Real Profile（APL=system_basic），用于需要高权限的应用 - --refresh-token 强制刷新 token 缓存（重新登录，连带刷新签名文件） - --refresh-signing 强制刷新签名文件缓存（重新申请，不重新登录） - -v, --verbose 显示调试日志 - --version 显示版本号 +hapsign auth [login|status] [--refresh] [--state-dir DIR] [--json] +hapsign devices [list] [--connected-only] [--json] +hapsign sign --hap HAP --serial SERIAL [签名选项] [--json] +hapsign install --hap SIGNED_HAP --serial SERIAL [--bundle-name NAME] [--json] +hapsign deploy --hap HAP --serial SERIAL [签名选项] [--json] + +sign / deploy 签名选项: + --bundle-name NAME 覆盖 HAP 中的包名 + --country CODE 华为账号国家码，默认 CN + --device-type TYPE 签名平台注册的设备类型码，默认 4 + --state-dir DIR Token 与默认签名材料根目录，默认 ~/.hapsign + --work-dir DIR 当前 bundle 签名材料目录，默认 / + --output-dir DIR 签名 HAP 输出目录，默认与 work-dir 相同 + --browser MODE system、system_controlled 或 playwright + --enable-capability 使用 Real Profile（APL=system_basic） + --refresh-token 强制浏览器认证，同时刷新签名材料 + --refresh-signing 只重新申请证书/Profile，复用有效 Token + -v, --verbose 将 DEBUG 日志写到 stderr 设备类型码: - 4 手机/平板（默认） + 4 手机/平板/2in1（默认） 2 穿戴设备 8 智慧屏 9 路由器 1 轻量级穿戴设备 + +退出码: + 0 命令成功 + 1 认证、签名、HDC 或安装运行失败 + 2 参数或输入 HAP 无效 + 130 用户取消 ``` +成功 JSON 至少包含 `ok=true` 和 `command`。`sign` / `deploy` 还包含 `input_hap`、 +`signed_hap`、`bundle_name`、`serial`、`input_signed` 和 `installed`；`devices list` +包含 `count`、`connected_count` 与 `targets`。失败 JSON 使用 +`{"ok":false,"command":"...","error":{"type":"...","message":"..."}}`。 +完整、随版本同步的帮助以 `hapsign --help` 和各子命令 `--help` 为准。 + ### 首次运行 会弹出浏览器窗口，打开华为登录页。手动输入账号密码登录，如果有验证码或二次验证也手动处理。登录成功后浏览器会自动关闭，后续自动完成签名和安装。 @@ -251,7 +292,7 @@ hapsign --hap [选项] 如果应用需要 `system_basic` 级别的 APL，加 `--enable-capability` 参数走 Real Provision 路径： ```bash -hapsign --hap app.hap --enable-capability +hapsign deploy --hap app.hap --serial --enable-capability ``` 此模式通过 `add.real.provision` API 创建 Real Profile（provisionType=1），对应 DevEco Studio 6.1+ 的 `enableCapability` 路径。需要应用已在 AGC（AppGallery Connect）注册且当前账号有访问权限，否则自动回退到 Test Profile。 @@ -269,8 +310,10 @@ HapSign/ └── signed_haps/ # 最新一个签名 HAP（可在设置中关闭） ``` -源码 CLI 默认保存在启动命令时所在目录的 -`signing_files//`；传入 `--work-dir` 可以指定其他目录。 +源码 CLI 默认保存在用户主目录的 `~/.hapsign//`；Windows 对应 +`%USERPROFILE%\.hapsign\\`。该默认值不依赖启动命令时的工作目录。 +传入 `--state-dir`、`--work-dir` 和 `--output-dir` 可以分别指定 Token/默认材料 +根目录、当前 bundle 材料目录和签名 HAP 输出目录。 程序目录必须可写，不建议把便携版解压到 `Program Files` 等受保护目录。 桌面版“设置”中还可以改为用户 `AppData Local` 或任意自定义目录。 @@ -290,15 +333,17 @@ HapSign/ ## 缓存策略 -同一天内不会重复登录或重复申请签名文件： +同一天内不会重复登录；签名文件只在 bundle 和目标 UDID 都相同时复用： -- **Token 缓存**：`signing_files/.token_cache.json`，当天复用，不重新登录 -- **签名文件缓存**：`signing_files/{bundle_name}/metadata.json`，当天复用，不重新申请证书/设备/Profile +- **Token 缓存**：`~/.hapsign/.token_cache.json`，当天可跨目标设备复用 +- **签名文件缓存**：`~/.hapsign/{bundle_name}/metadata.json`，当天仅为匹配的 + bundle 和设备 UDID 复用 - 跨天自动失效，重新走完整流程 - Token 失效时自动刷新，刷新失败才回退到重新登录 -这些文件包含明文敏感信息，不要上传、分享或放入云同步目录。共享电脑使用完毕后应删除 -`signing_files/`。详细说明见 [SECURITY.md](SECURITY.md)。 +Windows 使用当前用户 DPAPI 加密 Token；macOS/Linux 以权限 `0600` 的明文保存。 +签名材料与缓存都不要上传、分享或放入云同步目录。共享电脑使用完毕后应删除 +`~/.hapsign/`。详细说明见 [SECURITY.md](SECURITY.md)。 ## 限制 diff --git a/SECURITY.md b/SECURITY.md index 9bbf664..cc82819 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -10,12 +10,13 @@ vulnerability reporting；启用后请使用仓库 Security 页的“Report a vu ## 本地敏感数据 -hapsign 默认会在程序目录的 `signing_files/` 中保存当日 token 缓存、调试证书、 -Profile 和 `.p12` 密钥库，以避免重复登录和申请。这些文件已被 `.gitignore` -排除，但仍是本机敏感数据。Windows 上 token 缓存通过当前用户作用域的 DPAPI +桌面/便携版默认会在程序目录的 `signing_files/` 中保存当日 token 缓存、调试证书、 +Profile 和 `.p12` 密钥库；CLI 默认使用用户主目录的 `~/.hapsign/`（Windows 为 +`%USERPROFILE%\.hapsign\`）。这些文件已被 `.gitignore` 排除，但仍是本机敏感 +数据。Windows 上 token 缓存通过当前用户作用域的 DPAPI （CryptProtectData）静态加密后落盘，其他平台退化为受限权限（仅当前用户可读） 的明文存储，并在首次保存时打印告警；请勿把缓存目录放入云同步目录，在共享电脑 -上使用后应删除该目录。移动或分享便携目录前应先移除 `signing_files/`；桌面设置 +上使用后应删除对应目录。移动或分享便携目录前应先移除 `signing_files/`；桌面设置 可改为用户 AppData Local 或自定义目录；同样应按敏感数据目录保护。 程序目录的 `signed_haps/` 可能包含用户应用代码，移动或分享便携目录前也应检查； 可在设置中关闭保留签名 HAP。 diff --git a/docs/PACKAGING.md b/docs/PACKAGING.md index ae06a54..c787a3d 100644 --- a/docs/PACKAGING.md +++ b/docs/PACKAGING.md @@ -204,9 +204,10 @@ HapSign/ └── signed_haps/ # 最新一个签名 HAP，与材料目录设置无关 ``` -因此便携目录必须可写。源码 GUI 默认使用项目根目录，源码 CLI 默认使用当前工作 -目录下的 `signing_files//`。可以用 `HAPSIGN_DATA_DIR` 覆盖桌面版 -数据根目录，CLI 则可使用 `--work-dir`。已签名 HAP 会跳过签名流程，因此不会 +因此便携目录必须可写。源码 GUI 默认使用项目根目录，源码 CLI 默认使用用户主目录 +下的 `~/.hapsign//`。可以用 `HAPSIGN_DATA_DIR` 覆盖桌面版 +数据根目录，CLI 则可使用 `--state-dir`、`--work-dir` 和 `--output-dir`。已签名 +HAP 会跳过签名流程，因此不会 产生新的 `.p12`、`.cer`、`.p7b` 或签名后 HAP。 GUI 设置也可选择用户 `AppData Local` 或自定义签名目录。程序目录下的 diff --git a/hapsign/cli.py b/hapsign/cli.py index e97f794..3255fee 100644 --- a/hapsign/cli.py +++ b/hapsign/cli.py @@ -1,14 +1,32 @@ -"""hapsign 命令行入口。""" +"""Agent 友好的 HapSign 命令行入口。""" + +from __future__ import annotations import argparse +import contextlib +import io import json import logging +import os +import sys import zipfile from collections.abc import Sequence +from pathlib import Path from hapsign import __version__ +from hapsign.cancellation import OperationCancelled from hapsign.config import DEVICE_TYPE_PHONE -from hapsign.pipeline import SignPipeline +from hapsign.diagnostics import redact_sensitive_text +from hapsign.pipeline import SignPipeline, default_state_dir +from hapsign.signing.hap_inspect import is_hap_signed +from hapsign.signing.installer import Installer +from hapsign.token import secure_token_cache + +COMMANDS = {"auth", "devices", "sign", "install", "deploy"} +EXIT_OK = 0 +EXIT_OPERATION_FAILED = 1 +EXIT_USAGE = 2 +EXIT_CANCELLED = 130 def detect_bundle_name(hap_path: str) -> str: @@ -24,94 +42,533 @@ def detect_bundle_name(hap_path: str) -> str: raise ValueError("HAP 文件中未找到 module.json，无法提取 bundleName") -def build_parser() -> argparse.ArgumentParser: - """创建命令行参数解析器。""" - parser = argparse.ArgumentParser( - description="华为账号自动签名 + HAP 安装工具", - formatter_class=argparse.RawDescriptionHelpFormatter, - epilog="""\ -设备类型码: - 4 手机 (默认) - 2 穿戴设备 - 8 智慧屏 - 9 路由器 - 1 轻量级穿戴设备 +def _formatter(prog: str) -> argparse.HelpFormatter: + return argparse.RawDescriptionHelpFormatter(prog, max_help_position=30, width=100) -示例: - hapsign --hap app.hap - hapsign --hap app.hap --bundle-name com.example.myapp -""", + +def _add_output_options(parser: argparse.ArgumentParser) -> None: + parser.add_argument( + "--json", + dest="json_output", + action="store_true", + help="stdout 只输出单行 JSON；日志和诊断写到 stderr", ) + parser.add_argument("-v", "--verbose", action="store_true", help="输出 DEBUG 日志") + + +def _add_state_option(parser: argparse.ArgumentParser) -> None: parser.add_argument( - "--hap", - required=True, - help="HAP 文件路径（未签名则自动签名；已签名则直接安装）", + "--state-dir", + default=default_state_dir(), + help=( + "Token 与默认签名材料根目录；默认用户主目录 ~/.hapsign" + "（Windows 为 %%USERPROFILE%%\\.hapsign）。" + "Token 不会出现在 JSON 输出中" + ), ) + + +def _add_hap_identity_options(parser: argparse.ArgumentParser) -> None: + parser.add_argument("--hap", required=True, help="HAP 文件的绝对或相对路径") parser.add_argument( "--bundle-name", default=None, - help="应用包名（不传则从 HAP 内自动提取）", + help="覆盖 module.json 中的 bundleName；通常不需要指定", + ) + + +def _add_signing_options(parser: argparse.ArgumentParser) -> None: + _add_hap_identity_options(parser) + parser.add_argument( + "--serial", + required=True, + help="hdc list targets 返回的目标序列号；Profile 将绑定该设备", ) - parser.add_argument("--country", default="CN", help="国家码（默认 CN）") + parser.add_argument("--country", default="CN", help="华为账号国家码；默认 CN") parser.add_argument( "--device-type", default=DEVICE_TYPE_PHONE, - help="设备类型码（默认 4=手机，详见帮助底部）", + help="签名平台注册的设备类型码；默认 4（手机/平板/2in1）", ) + _add_state_option(parser) parser.add_argument( "--work-dir", default="", - help="工作目录（存储签名材料，默认 signing_files/{bundle_name}/）", + help="当前 bundle 的签名材料目录；默认 /", + ) + parser.add_argument( + "--output-dir", + default="", + help="签名 HAP 输出目录；默认与当前 bundle 的签名材料目录相同", + ) + parser.add_argument( + "--browser", + choices=("system", "system_controlled", "playwright"), + default="system", + help="首次认证使用的浏览器模式；CLI 默认 system", ) parser.add_argument( "--enable-capability", action="store_true", - help=( - "使用 Real Profile（APL=system_basic），用于需要高权限的应用。" - "需要应用已在 AGC 注册；未注册时自动回退到普通 Test Profile。" - ), + help="尝试使用 Real Profile（APL=system_basic）；需要已注册 AGC 应用", ) parser.add_argument( "--refresh-token", action="store_true", - help="强制重新登录，并连带刷新签名文件缓存。", + help="删除当日 Token 缓存并重新浏览器认证，同时刷新签名材料", ) parser.add_argument( "--refresh-signing", action="store_true", - help="强制重新申请证书、设备和 Profile，不重新登录。", + help="重新申请证书/Profile，但复用有效 Auth Token", ) - parser.add_argument("-v", "--verbose", action="store_true", help="显示调试日志") - parser.add_argument( - "--version", - action="version", - version=f"hapsign {__version__}", + _add_output_options(parser) + + +def build_parser() -> argparse.ArgumentParser: + """创建包含 Agent 子命令与明确退出码的参数解析器。""" + parser = argparse.ArgumentParser( + prog="hapsign", + description=( + "HarmonyOS HAP 账号认证、设备枚举、调试签名与安装 CLI。\n" + "面向 Agent：所有执行命令支持 --json，凭据永不写入 stdout。" + ), + formatter_class=_formatter, + epilog="""\ +典型 Agent 流程: + hapsign devices list --connected-only --json + hapsign auth status --json + hapsign auth --json + hapsign sign --hap app-unsigned.hap --serial --json + hapsign install --hap app-signed.hap --serial --json + hapsign deploy --hap app-unsigned.hap --serial --json + +兼容旧调用: + hapsign --hap app.hap --serial 等价于 hapsign deploy ... + +退出码: + 0 成功 + 1 运行失败（认证、签名、HDC 或安装失败） + 2 参数或输入文件无效 + 130 用户取消 +""", + ) + parser.add_argument("--version", action="version", version=f"hapsign {__version__}") + subparsers = parser.add_subparsers(dest="command", metavar="COMMAND") + + auth = subparsers.add_parser( + "auth", + help="登录或查看本地 Token 缓存状态；不检测设备、不签名、不安装", + description=( + "认证华为开发者账号并保存 Token。Auth Token 属于账号/Team，不绑定设备；" + "设备绑定发生在后续生成的 debug Profile 中。" + ), + formatter_class=_formatter, + epilog="""\ +示例: + hapsign auth --json # 有当日缓存则复用，否则打开浏览器 + hapsign auth --refresh --json # 强制重新浏览器认证 + hapsign auth status --json # 只检查本地缓存，不验证服务端有效性 +""", + ) + auth.add_argument( + "auth_action", + nargs="?", + choices=("login", "status"), + default="login", + help="login（默认）或 status", + ) + auth.add_argument("--country", default="CN", help="华为账号国家码；默认 CN") + auth.add_argument( + "--browser", + choices=("system", "system_controlled", "playwright"), + default="system", + help="认证浏览器模式；CLI 默认 system", + ) + auth.add_argument( + "--refresh", + action="store_true", + help="忽略并删除现有缓存，强制重新浏览器认证", + ) + _add_state_option(auth) + _add_output_options(auth) + + devices = subparsers.add_parser( + "devices", + help="列出 HDC targets，供 Agent 选择 --serial", + description=( + "列出 HDC 连接目标。JSON 会标记 connected、physical_candidate " + "和 likely_emulator；" + "serial 是 HDC 目标标识，不是 Profile 使用的 UDID。" + ), + formatter_class=_formatter, + epilog="""\ +示例: + hapsign devices list --json + hapsign devices list --connected-only --json +""", + ) + devices.add_argument( + "devices_action", + nargs="?", + choices=("list",), + default="list", + help="当前支持 list（默认）", + ) + devices.add_argument( + "--connected-only", action="store_true", help="只返回 Connected targets" ) + _add_output_options(devices) + + sign = subparsers.add_parser( + "sign", + help="为指定设备签名 HAP，但不安装", + description=( + "为 --serial 对应设备生成 debug Profile 并签名。" + "已有当日 Auth Token 会复用；" + "同一 bundle 切换设备时会自动丢弃不匹配的 Profile 缓存。" + ), + formatter_class=_formatter, + epilog="""\ +示例: + hapsign sign --hap app-unsigned.hap --serial --json + hapsign sign --hap app.hap --serial --output-dir ./signed --json +""", + ) + _add_signing_options(sign) + + install = subparsers.add_parser( + "install", + help="把已经签名的 HAP 覆盖安装到指定设备", + description=( + "只接受包含 Hap Signing Block 的已签名 HAP。安装使用 hdc -t " + "install -r，并在安装后通过 bm dump 确认 bundle。" + ), + formatter_class=_formatter, + epilog="""\ +示例: + hapsign install --hap app-signed.hap --serial --json +""", + ) + _add_hap_identity_options(install) + install.add_argument("--serial", required=True, help="目标 HDC 序列号") + _add_output_options(install) + + deploy = subparsers.add_parser( + "deploy", + help="自动认证、签名并安装；已签名 HAP 会直接安装", + description=( + "端到端命令。未签名 HAP：认证 → 证书/Profile → 签名 → 安装；" + "已签名 HAP：跳过认证和签名，直接覆盖安装。" + ), + formatter_class=_formatter, + epilog="""\ +示例: + hapsign deploy --hap app-unsigned.hap --serial --json + hapsign deploy --hap app-signed.hap --serial --json +""", + ) + _add_signing_options(deploy) return parser -def main(argv: Sequence[str] | None = None) -> int: - """运行命令行程序并返回进程退出码。""" - args = build_parser().parse_args(argv) +def _normalize_legacy_args(argv: Sequence[str]) -> list[str]: + """把旧 ``hapsign --hap ...`` 调用转换成 ``deploy`` 子命令。""" + normalized = list(argv) + if "--hap" in normalized and not any(item in COMMANDS for item in normalized[:1]): + normalized.insert(0, "deploy") + return normalized + + +def _command_from_args(argv: Sequence[str]) -> str: + return next((item for item in argv if item in COMMANDS), "unknown") + +def _parse_args( + parser: argparse.ArgumentParser, argv: Sequence[str] +) -> argparse.Namespace | int: + """解析参数；JSON 模式下把 argparse 错误转换为稳定 JSON。""" + if "--json" not in argv: + return parser.parse_args(argv) + + diagnostics = io.StringIO() + try: + with contextlib.redirect_stderr(diagnostics): + return parser.parse_args(argv) + except SystemExit as exc: + if exc.code == 0: + raise + lines = diagnostics.getvalue().strip().splitlines() + message = lines[-1].split("error:", 1)[-1].strip() if lines else "参数无效" + _failure( + _command_from_args(argv), + message, + True, + error_type="invalid_arguments", + ) + return EXIT_USAGE + + +def _configure_logging(verbose: bool) -> None: logging.basicConfig( - level=logging.DEBUG if args.verbose else logging.INFO, + level=logging.DEBUG if verbose else logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", datefmt="%H:%M:%S", ) - bundle_name = args.bundle_name or detect_bundle_name(args.hap) - if not args.bundle_name: - logging.info("自动检测到包名: %s", bundle_name) - pipeline = SignPipeline( - hap_path=args.hap, +def _emit(payload: dict[str, object], json_output: bool) -> None: + if json_output: + print(json.dumps(payload, ensure_ascii=False, sort_keys=True)) + return + if payload.get("ok"): + print(payload.get("message", "操作成功")) + for key in ( + "bundle_name", + "serial", + "signed_hap", + "cache_path", + "provision_type", + ): + value = payload.get(key) + if value: + print(f"{key}={value}") + return + error = payload.get("error") + if isinstance(error, dict): + print(f"错误: {error.get('message', '操作失败')}", file=sys.stderr) + + +def _failure( + command: str, + message: str, + json_output: bool, + *, + error_type: str = "operation_failed", +) -> None: + _emit( + { + "ok": False, + "command": command, + "error": {"type": error_type, "message": message}, + }, + json_output, + ) + + +def _resolve_hap(raw_path: str) -> Path: + path = Path(raw_path).expanduser().resolve() + if path.suffix.lower() != ".hap": + raise ValueError("--hap 必须指向 .hap 文件") + if not path.is_file(): + raise ValueError(f"HAP 文件不存在: {path}") + return path + + +def _auth_pipeline(args: argparse.Namespace) -> SignPipeline: + state_dir = str(Path(args.state_dir).expanduser().resolve()) + return SignPipeline( + hap_path="", + bundle_name="__auth__", + work_dir=state_dir, + state_dir=state_dir, + country=args.country, + browser_mode=args.browser, + keep_signed_hap=False, + install_after_sign=False, + ) + + +def _cache_format(cache_path: str) -> str: + try: + raw = Path(cache_path).read_bytes() + except OSError: + return "missing" + if secure_token_cache.is_encrypted(raw): + return "dpapi-v1" + return "plaintext-json-0600" if os.name != "nt" else "legacy-plaintext" + + +def _run_auth(args: argparse.Namespace) -> int: + pipeline = _auth_pipeline(args) + if args.auth_action == "status": + status = pipeline.auth_status() + status.update( + { + "ok": True, + "command": "auth status", + "message": ( + "发现当日认证缓存" + if status["authenticated"] + else "没有可用的当日认证缓存" + ), + "cache_format": _cache_format(str(status["cache_path"])), + } + ) + _emit(status, args.json_output) + return EXIT_OK + + result = pipeline.authenticate(force_refresh=args.refresh) + status = pipeline.auth_status() + result.update( + { + "ok": True, + "command": "auth", + "message": "认证缓存可用", + "cache_path": str(status["cache_path"]), + "cache_format": _cache_format(str(status["cache_path"])), + } + ) + _emit(result, args.json_output) + return EXIT_OK + + +def _run_devices(args: argparse.Namespace) -> int: + with Installer() as installer: + targets = installer.list_targets(connected_only=args.connected_only) + connected_count = sum(bool(target["connected"]) for target in targets) + payload: dict[str, object] = { + "ok": True, + "command": "devices list", + "message": f"发现 {len(targets)} 个 HDC target", + "count": len(targets), + "connected_count": connected_count, + "targets": targets, + } + _emit(payload, args.json_output) + if not args.json_output: + for target in targets: + print("{serial}\t{transport}\t{status}\t{host}".format(**target).rstrip()) + return EXIT_OK + + +def _build_sign_pipeline( + args: argparse.Namespace, + hap_path: Path, + bundle_name: str, + *, + install_after_sign: bool, +) -> SignPipeline: + return SignPipeline( + hap_path=str(hap_path), bundle_name=bundle_name, country=args.country, device_type=args.device_type, + serial=args.serial, work_dir=args.work_dir, + state_dir=args.state_dir, enable_capability=args.enable_capability, force_refresh_token=args.refresh_token, force_refresh_signing=args.refresh_signing, + browser_mode=args.browser, + signed_output_dir=args.output_dir, + keep_signed_hap=True, + install_after_sign=install_after_sign, + ) + + +def _inspect_installed_bundle(serial: str, bundle_name: str) -> dict[str, str]: + with Installer(serial=serial) as installer: + bundle = installer.inspect_bundle(bundle_name) + if bundle is None: + raise RuntimeError("安装命令结束后 bm dump 未确认目标 bundle") + return bundle + + +def _run_sign_or_deploy(args: argparse.Namespace) -> int: + hap_path = _resolve_hap(args.hap) + bundle_name = args.bundle_name or detect_bundle_name(str(hap_path)) + input_signed = is_hap_signed(hap_path) + deploy = args.command == "deploy" + pipeline = _build_sign_pipeline( + args, + hap_path, + bundle_name, + install_after_sign=deploy, ) - return 0 if pipeline.run() else 1 + if not pipeline.run(): + message = getattr(pipeline, "last_error", "") or ( + "流程未完成；请查看 stderr 日志" + ) + _failure(args.command, message, args.json_output) + return EXIT_OPERATION_FAILED + + signed_hap = str(Path(pipeline.signed_hap_path).resolve()) + payload: dict[str, object] = { + "ok": True, + "command": args.command, + "message": "签名并安装成功" if deploy else "签名成功", + "bundle_name": bundle_name, + "serial": args.serial, + "input_hap": str(hap_path), + "input_signed": input_signed, + "signed_hap": signed_hap, + "installed": deploy, + } + if deploy: + bundle = _inspect_installed_bundle(args.serial, bundle_name) + payload.update(bundle) + _emit(payload, args.json_output) + return EXIT_OK + + +def _run_install(args: argparse.Namespace) -> int: + hap_path = _resolve_hap(args.hap) + if not is_hap_signed(hap_path): + raise ValueError("install 只接受已签名 HAP；请先使用 sign 或直接使用 deploy") + bundle_name = args.bundle_name or detect_bundle_name(str(hap_path)) + with Installer(serial=args.serial) as installer: + installer.install(str(hap_path)) + bundle = installer.inspect_bundle(bundle_name) + if bundle is None: + raise RuntimeError("安装命令结束后 bm dump 未确认目标 bundle") + payload: dict[str, object] = { + "ok": True, + "command": "install", + "message": "安装成功", + "bundle_name": bundle_name, + "serial": args.serial, + "signed_hap": str(hap_path), + "installed": True, + } + payload.update(bundle) + _emit(payload, args.json_output) + return EXIT_OK + + +def main(argv: Sequence[str] | None = None) -> int: + """运行 CLI 并返回稳定退出码；JSON 模式永不输出凭据。""" + raw_args = list(sys.argv[1:] if argv is None else argv) + parser = build_parser() + normalized_args = _normalize_legacy_args(raw_args) + parsed = _parse_args(parser, normalized_args) + if isinstance(parsed, int): + return parsed + args = parsed + if args.command is None: + parser.print_help() + return EXIT_USAGE + _configure_logging(args.verbose) + + try: + if args.command == "auth": + return _run_auth(args) + if args.command == "devices": + return _run_devices(args) + if args.command in {"sign", "deploy"}: + return _run_sign_or_deploy(args) + if args.command == "install": + return _run_install(args) + raise ValueError(f"未知命令: {args.command}") + except OperationCancelled: + _failure(args.command, "操作已取消", args.json_output, error_type="cancelled") + return EXIT_CANCELLED + except (ValueError, FileNotFoundError, zipfile.BadZipFile) as exc: + message = str(redact_sensitive_text(exc)) + _failure(args.command, message, args.json_output, error_type="invalid_input") + return EXIT_USAGE + except Exception as exc: + message = str(redact_sensitive_text(exc)) + logging.getLogger(__name__).error("CLI 操作失败: %s", message) + _failure(args.command, message, args.json_output) + return EXIT_OPERATION_FAILED diff --git a/hapsign/pipeline.py b/hapsign/pipeline.py index 40738b8..26354fe 100644 --- a/hapsign/pipeline.py +++ b/hapsign/pipeline.py @@ -1,9 +1,9 @@ """登录、token 交换、签名材料申请、HAP 签名和安装的全流程编排。 双重缓存策略（同一天内复用，避免反复登录和申请）： - 1. Token 缓存：``signing_files/.token_cache.json`` 存储当天登录的 token 信息， + 1. Token 缓存：``~/.hapsign/.token_cache.json`` 存储当天登录的 token 信息， 同账号同一天内复用，不重新登录。 - 2. 签名文件缓存：``signing_files/{bundle_name}/metadata.json`` 存储当天申请的 + 2. 签名文件缓存：``~/.hapsign/{bundle_name}/metadata.json`` 存储当天申请的 签名文件路径，同一天内复用，不重新申请证书/设备/Profile。 缓存失效场景： @@ -21,6 +21,7 @@ import zipfile from collections.abc import Callable from datetime import date +from pathlib import Path from hapsign.api.capability_api import CapabilityAPI from hapsign.api.cert_api import CertAPI @@ -46,8 +47,18 @@ logger = logging.getLogger(__name__) -# 签名文件根目录（相对于项目根） -SIGNING_FILES_DIR = "signing_files" +# 默认签名状态目录。Path.home() 在 macOS/Linux 解析为 $HOME，在 Windows +# 解析为 %USERPROFILE%，因此不会依赖调用命令时的工作目录。 +DEFAULT_STATE_DIR_NAME = ".hapsign" + + +def default_state_dir() -> str: + """返回跨平台的默认 Token 与签名材料目录。""" + return str((Path.home() / DEFAULT_STATE_DIR_NAME).resolve()) + + +# 保留旧常量名，避免外部调用方导入时报错；新代码应调用 default_state_dir()。 +SIGNING_FILES_DIR = default_state_dir() SIGNED_HAP_MANIFEST = ".hapsign-signed-haps.json" @@ -83,11 +94,14 @@ def __init__( keep_signed_hap: bool = True, cancel_event: threading.Event | None = None, progress_callback: Callable[[int, str], None] | None = None, + serial: str | None = None, + install_after_sign: bool = True, ): self.hap_path = hap_path self.bundle_name = bundle_name self.country = country self.device_type = device_type + self.serial = serial self.enable_capability = enable_capability self.force_refresh_token = force_refresh_token self.force_refresh_signing = force_refresh_signing @@ -96,13 +110,22 @@ def __init__( self._temporary_signed_dir: tempfile.TemporaryDirectory | None = None self.cancel_event = cancel_event self.progress_callback = progress_callback - self.state_dir = state_dir or SIGNING_FILES_DIR + self.state_dir = ( + os.path.abspath(os.path.expanduser(state_dir)) + if state_dir + else default_state_dir() + ) + self.install_after_sign = install_after_sign if work_dir: - self.work_dir = work_dir + self.work_dir = os.path.abspath(os.path.expanduser(work_dir)) else: self.work_dir = os.path.join(self.state_dir, bundle_name) - self.signed_output_dir = signed_output_dir or self.work_dir - os.makedirs(self.work_dir, exist_ok=True) + self.signed_output_dir = ( + os.path.abspath(os.path.expanduser(signed_output_dir)) + if signed_output_dir + else self.work_dir + ) + os.makedirs(self.work_dir, mode=0o700, exist_ok=True) self.keystore_password = KEYSTORE_PASSWORD self._metadata_path = os.path.join(self.work_dir, "metadata.json") self._token_cache_path = os.path.join(self.state_dir, ".token_cache.json") @@ -128,6 +151,7 @@ def __init__( self._csr_path = "" self._csr_content = "" self._signed_hap_path = "" + self._last_error = "" def _check_cancelled(self) -> None: raise_if_cancelled(self.cancel_event) @@ -141,9 +165,9 @@ def _emit_progress(self, value: int, label: str) -> None: def _load_token_cache(self) -> dict | None: """加载当天的 token 缓存。 - 条件：缓存存在、creation_date 是今天；缓存可能是 DPAPI 加密格式或旧版 - 明文 JSON。明文缓存首次读取时安全迁移为加密格式；解密失败视为无缓存， - 回退重新登录。 + 条件：缓存存在、creation_date 是今天；缓存可能是 Windows DPAPI 加密格式 + 或非 Windows 平台的 0600 明文 JSON。Windows 上的旧版明文缓存首次读取时 + 安全迁移为加密格式；解密失败视为无缓存，回退重新登录。 """ if not os.path.exists(self._token_cache_path): return None @@ -173,7 +197,9 @@ def _load_token_cache(self) -> dict | None: cache = json.loads(raw) except (json.JSONDecodeError, UnicodeDecodeError, TypeError): return None - legacy = True + # 非 Windows 平台没有 DPAPI，0600 明文就是当前格式；只有 Windows + # 才把明文视为旧格式并迁移，避免 macOS 每次读取都重复写盘和告警。 + legacy = os.name == "nt" if not isinstance(cache, dict): logger.warning("[cache] token 缓存格式无效，将重新登录") @@ -288,6 +314,11 @@ def _load_cached_metadata(self) -> dict | None: logger.info("[cache] 签名文件创建于 %s，非今日", meta.get("creation_date")) return None + cached_udid = str(meta.get("udid", "")) + if self._udid and cached_udid != self._udid: + logger.info("[cache] 签名 Profile 属于其他设备，重新申请") + return None + for key in ("p12_path", "cer_path", "p7b_path"): path = meta.get(key, "") if not path or not os.path.exists(path): @@ -329,7 +360,8 @@ def run(self) -> bool: self._emit_progress(2, "正在准备") result = self._run_pipeline() if result: - self._emit_progress(100, "安装完成") + final_label = "安装完成" if self.install_after_sign else "签名完成" + self._emit_progress(100, final_label) return result finally: self._close_installer() @@ -353,9 +385,11 @@ def _run_pipeline(self) -> bool: # 已签名包：跳过登录 / 申请 / 签名，直接安装原文件 if is_hap_signed(self.hap_path): - logger.info("[sign] 检测到已签名 HAP，跳过签名流程，直接安装") + logger.info("[sign] 检测到已签名 HAP，跳过签名流程") self._signed_hap_path = self.hap_path - steps = [("安装 hap 到设备", self._step_install)] + steps = [] + if self.install_after_sign: + steps.append(("安装 hap 到设备", self._step_install)) return self._run_steps(steps) # 强制刷新 token 时连带刷新签名文件（仅本次运行，不改写实例属性） @@ -378,8 +412,9 @@ def _run_pipeline(self) -> bool: self._p7b_path = signing_cached["p7b_path"] steps = [ ("签名 hap", self._step_sign_hap), - ("安装 hap 到设备", self._step_install), ] + if self.install_after_sign: + steps.append(("安装 hap 到设备", self._step_install)) else: # 需要申请签名文件，先确保有 token if not self.force_refresh_token: @@ -412,8 +447,9 @@ def _run_pipeline(self) -> bool: ("注册调试设备", self._step_register_device), ("创建调试 Profile", self._step_create_provision), ("签名 hap", self._step_sign_hap), - ("安装 hap 到设备", self._step_install), ] + if self.install_after_sign: + steps.append(("安装 hap 到设备", self._step_install)) if self.enable_capability: steps.insert(0, ("查询应用信息", self._step_get_app_info)) @@ -456,6 +492,7 @@ def _run_steps(self, steps: list) -> bool: except OperationCancelled: raise except Exception as retry_err: + self._last_error = f"{name}: {redact_sensitive_text(retry_err)}" logger.error( "x %s failed after retry: %s", name, @@ -470,6 +507,7 @@ def _run_steps(self, steps: list) -> bool: except OperationCancelled: raise except Exception as e: + self._last_error = f"{name}: {redact_sensitive_text(e)}" logger.error("x %s failed: %s", name, redact_sensitive_text(e)) logger.debug( "%s 失败调用栈:\n%s", @@ -512,6 +550,48 @@ def _step_exchange_token(self) -> None: # 保存 token 缓存，供同一天内复用 self._save_token_cache() + def authenticate(self, force_refresh: bool = False) -> dict[str, object]: + """确保账号认证缓存可用，不执行设备检测、签名或安装。""" + if force_refresh: + self._clear_token_cache() + cache = self._load_token_cache() + if cache is not None: + self._init_client_from_cache(cache) + return { + "authenticated": True, + "from_cache": True, + "creation_date": cache.get("creation_date", ""), + } + + self._step_login() + self._step_exchange_token() + return { + "authenticated": True, + "from_cache": False, + "creation_date": date.today().isoformat(), + } + + def auth_status(self) -> dict[str, object]: + """返回不含凭据内容的本地认证缓存状态。""" + cache = self._load_token_cache() + return { + "authenticated": cache is not None, + "cached": cache is not None, + "creation_date": cache.get("creation_date", "") if cache else "", + "cache_path": os.path.abspath(self._token_cache_path), + "online_verified": False, + } + + @property + def signed_hap_path(self) -> str: + """返回本次运行选择或生成的签名 HAP 路径。""" + return self._signed_hap_path + + @property + def last_error(self) -> str: + """返回适合 CLI 展示的最近一次脱敏流程错误。""" + return self._last_error + def _step_get_app_info(self) -> None: """查询应用简要信息，获取 appId 和 projectId。 @@ -849,7 +929,7 @@ def _step_install(self) -> None: self._get_installer().install(self._signed_hap_path) def _step_check_device(self) -> None: - """确认有且仅有一台已授权、可通过 HDC 访问的设备。""" + """读取显式目标或 HDC 默认目标的 UDID，确认设备可访问。""" self._udid = self._get_installer().get_udid() logger.info("已检测到可用设备（UDID 尾号 %s）", self._udid[-6:]) @@ -857,7 +937,10 @@ def _step_check_device(self) -> None: def _get_installer(self) -> Installer: if self._installer is None: - self._installer = Installer(cancel_event=self.cancel_event) + self._installer = Installer( + cancel_event=self.cancel_event, + serial=self.serial, + ) return self._installer def _cleanup_temporary_signed_hap(self) -> None: diff --git a/hapsign/signing/installer.py b/hapsign/signing/installer.py index cc5308e..b861985 100644 --- a/hapsign/signing/installer.py +++ b/hapsign/signing/installer.py @@ -156,14 +156,27 @@ def _process_start_time_posix(pid: int) -> float | None: class Installer: """使用 hdc 获取设备 UDID 并安装 hap 包。""" - def __init__(self, cancel_event: threading.Event | None = None) -> None: + def __init__( + self, + cancel_event: threading.Event | None = None, + serial: str | None = None, + ) -> None: self._hdc = config.HDC_PATH self.cancel_event = cancel_event + self.serial = serial # 本任务确认创建、close 时应清理的 HDC server 监听 PID；None 表示不归属 self._owned_server_pid: int | None = None self._server_checked = False self._closed = False + def _device_command(self, *args: str) -> list[str]: + """构建设备命令；指定 serial 时显式选择目标，避免 HDC 隐式歧义。""" + command = [self._hdc] + if self.serial: + command.extend(["-t", self.serial]) + command.extend(args) + return command + def __enter__(self) -> Installer: return self @@ -296,8 +309,9 @@ def get_udid(self) -> str: """ self._ensure_server() commands = [ - [self._hdc, "shell", "bm", "get", "-u"], - [self._hdc, "shell", "param", "get", "const.product.udid"], + self._device_command("shell", "bm", "get", "--udid"), + self._device_command("shell", "bm", "get", "-u"), + self._device_command("shell", "param", "get", "const.product.udid"), ] for cmd in commands: try: @@ -319,6 +333,78 @@ def get_udid(self) -> str: "未检测到可用设备：请确认设备已连接、已授权 USB 调试，且当前只连接一台设备" ) + def list_targets(self, connected_only: bool = False) -> list[dict[str, object]]: + """列出 HDC targets，返回适合 CLI JSON 输出的非敏感设备信息。""" + self._ensure_server() + result = run_process( + [self._hdc, "list", "targets", "-v"], + capture_output=True, + text=True, + timeout=15, + cancel_event=self.cancel_event, + ) + if result.returncode != 0: + output = (result.stderr or result.stdout or "").strip() + raise RuntimeError(f"hdc list targets 失败: {output}") + + targets: list[dict[str, object]] = [] + for raw_line in result.stdout.splitlines(): + parts = raw_line.strip().split() + if not parts or parts[0].startswith("["): + continue + serial = parts[0] + transport = parts[1] if len(parts) > 1 else "" + status = parts[2] if len(parts) > 2 else "" + host = " ".join(parts[3:]) if len(parts) > 3 else "" + connected = status.lower() == "connected" + usb = transport.upper() == "USB" + localhost = serial.startswith(("127.0.0.1:", "localhost:")) + if connected_only and not connected: + continue + targets.append( + { + "serial": serial, + "transport": transport, + "status": status, + "host": host, + "connected": connected, + "usb": usb, + "localhost": localhost, + "physical_candidate": usb and connected, + "likely_emulator": localhost and not usb, + } + ) + return targets + + def inspect_bundle(self, bundle_name: str) -> dict[str, str] | None: + """查询已安装 bundle；不存在或输出不匹配时返回 None。""" + self._ensure_server() + result = run_process( + self._device_command("shell", "bm", "dump", "-n", bundle_name), + capture_output=True, + text=True, + timeout=20, + cancel_event=self.cancel_event, + ) + output = (result.stdout or "") + (result.stderr or "") + bundle_match = re.search(r'"bundleName"\s*:\s*"([^"]+)"', output) + if ( + result.returncode != 0 + or bundle_match is None + or bundle_match.group(1) != bundle_name + ): + return None + provision_match = re.search( + r'"(?:appProvisionType|provisionType)"\s*:\s*"([^"]+)"', + output, + ) + version_match = re.search(r'"versionName"\s*:\s*"([^"]+)"', output) + return { + "bundle_name": bundle_name, + "provision_type": provision_match.group(1) if provision_match else "", + "version_name": version_match.group(1) if version_match else "", + } + def install(self, hap_path: str) -> bool: """使用 hdc install 安装 hap 包到已连接设备。 @@ -332,7 +418,7 @@ def install(self, hap_path: str) -> bool: RuntimeError: hdc install 执行失败时抛出。 """ self._ensure_server() - cmd = [self._hdc, "install", hap_path] + cmd = self._device_command("install", "-r", hap_path) result = run_process( cmd, capture_output=True, @@ -343,7 +429,12 @@ def install(self, hap_path: str) -> bool: output = (result.stdout or "") + (result.stderr or "") # hdc install 即使失败也可能返回 0，仅按真实失败标记识别： # 非零退出码、[Fail] 状态行、INSTALL_FAILED_ 前缀或明确的 error: 行。 - failed = result.returncode != 0 or bool(_FATAL_INSTALL_MARKERS.search(output)) + succeeded = "install bundle successfully" in output.lower() + failed = ( + result.returncode != 0 + or bool(_FATAL_INSTALL_MARKERS.search(output)) + or not succeeded + ) if failed: raise RuntimeError( f"hdc install 失败 (code={result.returncode}): {output.strip()}" diff --git a/sign_install.bat b/sign_install.bat index 47211f4..4aede45 100644 --- a/sign_install.bat +++ b/sign_install.bat @@ -18,10 +18,11 @@ if defined HAPSIGN_EXTRA_ARGS set EXTRA_ARGS=%EXTRA_ARGS% %HAPSIGN_EXTRA_ARGS% if "%~1"=="" ( echo Usage: drag a .hap file onto this script, or: - echo sign_install.bat path\to\app-unsigned.hap + echo sign_install.bat path\to\app-unsigned.hap device-serial echo. echo Environment variables: echo HAPSIGN_PYTHON Python path + echo HAPSIGN_SERIAL Required HDC target serial echo HAPSIGN_REFRESH_TOKEN=1 Force refresh token cache echo HAPSIGN_REFRESH_SIGNING=1 Force refresh signing files echo HAPSIGN_EXTRA_ARGS Extra CLI args ^(e.g. --enable-capability^) @@ -29,14 +30,28 @@ if "%~1"=="" ( exit /b 1 ) +set TARGET_SERIAL=%HAPSIGN_SERIAL% +if not "%~2"=="" set TARGET_SERIAL=%~2 +if not defined TARGET_SERIAL ( + echo Error: device serial is required. + echo Run the following command, then pass a Connected USB serial as argument 2 + echo or set HAPSIGN_SERIAL: + echo. + "%PYTHON%" -u main.py devices list + echo. + pause + exit /b 2 +) + echo ======================================== echo hapsign - Auto Sign ^& Install echo ======================================== echo Hap: %~1 +echo Serial: %TARGET_SERIAL% if defined EXTRA_ARGS echo Extra: %EXTRA_ARGS% echo. -"%PYTHON%" -u main.py --hap "%~1" %EXTRA_ARGS% +"%PYTHON%" -u main.py deploy --hap "%~1" --serial "%TARGET_SERIAL%" %EXTRA_ARGS% echo. if %ERRORLEVEL% equ 0 ( diff --git a/tests/test_cli.py b/tests/test_cli.py index 0d75bc7..7486287 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -1,4 +1,4 @@ -"""命令行入口测试。""" +"""命令行入口与 Agent JSON 协议测试。""" import json import zipfile @@ -8,14 +8,15 @@ from hapsign import cli -def _write_hap(path, module_data: dict) -> None: +def _write_hap(path, module_data: dict | None = None) -> None: + data = module_data or {"app": {"bundleName": "com.example.app"}} with zipfile.ZipFile(path, "w") as archive: - archive.writestr("module.json", json.dumps(module_data)) + archive.writestr("module.json", json.dumps(data)) def test_detect_bundle_name(tmp_path) -> None: hap_path = tmp_path / "app.hap" - _write_hap(hap_path, {"app": {"bundleName": "com.example.app"}}) + _write_hap(hap_path) assert cli.detect_bundle_name(str(hap_path)) == "com.example.app" @@ -37,21 +38,327 @@ def test_detect_bundle_name_requires_bundle_name(tmp_path) -> None: cli.detect_bundle_name(str(hap_path)) +def test_legacy_hap_invocation_maps_to_deploy() -> None: + assert cli._normalize_legacy_args( + ["--hap", "app.hap", "--serial", "device-serial"] + ) == ["deploy", "--hap", "app.hap", "--serial", "device-serial"] + + +def test_parser_uses_home_state_dir_by_default(monkeypatch, tmp_path) -> None: + expected = str(tmp_path / ".hapsign") + monkeypatch.setattr(cli, "default_state_dir", lambda: expected) + + args = cli.build_parser().parse_args(["auth", "status"]) + + assert args.state_dir == expected + + +def test_auth_help_renders_windows_home_example(capsys) -> None: + parser = cli.build_parser() + + with pytest.raises(SystemExit, match="0"): + parser.parse_args(["auth", "--help"]) + + help_text = capsys.readouterr().out + assert "%USERPROFILE%\\.hapsign" in help_text + + @pytest.mark.parametrize(("pipeline_result", "exit_code"), [(True, 0), (False, 1)]) -def test_main_returns_pipeline_result(monkeypatch, pipeline_result, exit_code) -> None: +def test_deploy_json_returns_pipeline_result( + monkeypatch, tmp_path, capsys, pipeline_result, exit_code +) -> None: + hap_path = tmp_path / "example.hap" + _write_hap(hap_path) captured = {} class FakePipeline: def __init__(self, **kwargs): captured.update(kwargs) + self.signed_hap_path = str(hap_path) + self.last_error = "检测设备连接: device unavailable" def run(self): return pipeline_result monkeypatch.setattr(cli, "SignPipeline", FakePipeline) + monkeypatch.setattr(cli, "is_hap_signed", lambda _path: False) + monkeypatch.setattr( + cli, + "_inspect_installed_bundle", + lambda *_args: { + "bundle_name": "com.example.app", + "provision_type": "debug", + "version_name": "1.0.0", + }, + ) - result = cli.main(["--hap", "example.hap", "--bundle-name", "com.example.app"]) + result = cli.main( + [ + "deploy", + "--hap", + str(hap_path), + "--serial", + "device-serial", + "--json", + ] + ) assert result == exit_code - assert captured["hap_path"] == "example.hap" + payload = json.loads(capsys.readouterr().out) + assert payload["ok"] is pipeline_result + assert payload["command"] == "deploy" + assert captured["hap_path"] == str(hap_path) assert captured["bundle_name"] == "com.example.app" + assert captured["serial"] == "device-serial" + assert captured["install_after_sign"] is True + if pipeline_result: + assert payload["installed"] is True + assert payload["signed_hap"] == str(hap_path) + else: + assert payload["error"]["message"] == ("检测设备连接: device unavailable") + + +def test_sign_json_does_not_install(monkeypatch, tmp_path, capsys) -> None: + hap_path = tmp_path / "unsigned.hap" + signed_path = tmp_path / "unsigned_signed.hap" + _write_hap(hap_path) + captured = {} + + class FakePipeline: + def __init__(self, **kwargs): + captured.update(kwargs) + self.signed_hap_path = str(signed_path) + + def run(self): + return True + + monkeypatch.setattr(cli, "SignPipeline", FakePipeline) + monkeypatch.setattr(cli, "is_hap_signed", lambda _path: False) + monkeypatch.setattr( + cli, + "_inspect_installed_bundle", + lambda *_args: pytest.fail("sign 不应检查已安装 bundle"), + ) + + assert ( + cli.main( + [ + "sign", + "--hap", + str(hap_path), + "--serial", + "device-serial", + "--json", + ] + ) + == 0 + ) + + payload = json.loads(capsys.readouterr().out) + assert payload["ok"] is True + assert payload["installed"] is False + assert payload["signed_hap"] == str(signed_path) + assert captured["install_after_sign"] is False + + +def test_auth_status_json_never_emits_token(monkeypatch, tmp_path, capsys) -> None: + cache_path = tmp_path / ".token_cache.json" + cache_path.write_text("{}", encoding="utf-8") + + class FakePipeline: + def __init__(self, **_kwargs): + pass + + def auth_status(self): + return { + "authenticated": True, + "cached": True, + "creation_date": "2026-08-28", + "cache_path": str(cache_path), + "online_verified": False, + } + + monkeypatch.setattr(cli, "SignPipeline", FakePipeline) + + assert cli.main(["auth", "status", "--json"]) == 0 + + output = capsys.readouterr().out + payload = json.loads(output) + assert payload["authenticated"] is True + assert payload["online_verified"] is False + assert "access_token" not in output + assert "refresh_token" not in output + assert "jwt_token" not in output + + +def test_auth_login_reports_cache_reuse(monkeypatch, tmp_path, capsys) -> None: + cache_path = tmp_path / ".token_cache.json" + cache_path.write_text("{}", encoding="utf-8") + + class FakePipeline: + def __init__(self, **_kwargs): + pass + + def authenticate(self, force_refresh=False): + assert force_refresh is False + return { + "authenticated": True, + "from_cache": True, + "creation_date": "2026-08-28", + } + + def auth_status(self): + return {"cache_path": str(cache_path)} + + monkeypatch.setattr(cli, "SignPipeline", FakePipeline) + + assert cli.main(["auth", "--json"]) == 0 + + payload = json.loads(capsys.readouterr().out) + assert payload["ok"] is True + assert payload["from_cache"] is True + + +def test_devices_list_json(monkeypatch, capsys) -> None: + targets = [ + { + "serial": "device-serial", + "transport": "USB", + "status": "Connected", + "host": "localhost", + "connected": True, + "usb": True, + "localhost": False, + "physical_candidate": True, + "likely_emulator": False, + } + ] + + class FakeInstaller: + def __enter__(self): + return self + + def __exit__(self, *_args): + return None + + def list_targets(self, connected_only=False): + assert connected_only is True + return targets + + monkeypatch.setattr(cli, "Installer", FakeInstaller) + + assert cli.main(["devices", "list", "--connected-only", "--json"]) == 0 + + payload = json.loads(capsys.readouterr().out) + assert payload["count"] == 1 + assert payload["connected_count"] == 1 + assert payload["targets"] == targets + + +def test_json_argument_error_has_stable_exit_code(capsys) -> None: + assert cli.main(["sign", "--hap", "app.hap", "--json"]) == 2 + + payload = json.loads(capsys.readouterr().out) + assert payload["ok"] is False + assert payload["command"] == "sign" + assert payload["error"]["type"] == "invalid_arguments" + assert "--serial" in payload["error"]["message"] + + +def test_missing_hap_json_is_invalid_input(capsys, tmp_path) -> None: + missing = tmp_path / "missing.hap" + + assert ( + cli.main( + [ + "sign", + "--hap", + str(missing), + "--serial", + "device-serial", + "--json", + ] + ) + == 2 + ) + + payload = json.loads(capsys.readouterr().out) + assert payload["ok"] is False + assert payload["error"]["type"] == "invalid_input" + + +def test_install_signed_hap_with_explicit_serial(monkeypatch, tmp_path, capsys) -> None: + hap_path = tmp_path / "signed.hap" + _write_hap(hap_path) + calls = [] + + class FakeInstaller: + def __init__(self, serial=None): + calls.append(("serial", serial)) + + def __enter__(self): + return self + + def __exit__(self, *_args): + return None + + def install(self, path): + calls.append(("install", path)) + + def inspect_bundle(self, bundle_name): + calls.append(("inspect", bundle_name)) + return { + "bundle_name": bundle_name, + "provision_type": "debug", + "version_name": "1.0.0", + } + + monkeypatch.setattr(cli, "Installer", FakeInstaller) + monkeypatch.setattr(cli, "is_hap_signed", lambda _path: True) + + assert ( + cli.main( + [ + "install", + "--hap", + str(hap_path), + "--serial", + "device-serial", + "--json", + ] + ) + == 0 + ) + + payload = json.loads(capsys.readouterr().out) + assert payload["ok"] is True + assert payload["installed"] is True + assert calls == [ + ("serial", "device-serial"), + ("install", str(hap_path)), + ("inspect", "com.example.app"), + ] + + +def test_install_rejects_unsigned_hap(monkeypatch, tmp_path, capsys) -> None: + hap_path = tmp_path / "unsigned.hap" + _write_hap(hap_path) + monkeypatch.setattr(cli, "is_hap_signed", lambda _path: False) + + assert ( + cli.main( + [ + "install", + "--hap", + str(hap_path), + "--serial", + "device-serial", + "--json", + ] + ) + == 2 + ) + + payload = json.loads(capsys.readouterr().out) + assert payload["error"]["type"] == "invalid_input" + assert "deploy" in payload["error"]["message"] diff --git a/tests/test_config_paths.py b/tests/test_config_paths.py index 5adc5e6..4dcc3fa 100644 --- a/tests/test_config_paths.py +++ b/tests/test_config_paths.py @@ -2,6 +2,7 @@ import os +from hapsign import pipeline from hapsign.config import default_deveco_home, resolve_sdk_paths @@ -11,6 +12,17 @@ def test_default_deveco_home_by_platform() -> None: assert default_deveco_home("linux") == "/opt/DevEco-Studio" +def test_default_state_dir_is_hidden_folder_under_home(tmp_path, monkeypatch) -> None: + class FakePath: + @staticmethod + def home(): + return tmp_path + + monkeypatch.setattr(pipeline, "Path", FakePath) + + assert pipeline.default_state_dir() == str((tmp_path / ".hapsign").resolve()) + + def test_resolve_sdk_paths_darwin() -> None: home = "/Applications/DevEco-Studio.app/Contents" java, hap_sign, hdc, keytool = resolve_sdk_paths(home, "darwin") diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index 68e3cb8..55a8f3e 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -21,6 +21,15 @@ from hapsign.token import secure_token_cache +@pytest.fixture(autouse=True) +def _isolate_default_state_dir(tmp_path, monkeypatch) -> None: + """禁止无 state-dir 的测试接触真实用户主目录。""" + monkeypatch.setattr( + "hapsign.pipeline.default_state_dir", + lambda: str(tmp_path / "signing_files"), + ) + + def _pipeline(tmp_path, monkeypatch) -> SignPipeline: monkeypatch.chdir(tmp_path) return SignPipeline( @@ -239,6 +248,67 @@ def test_token_cache_migrates_legacy_plaintext(tmp_path, monkeypatch) -> None: assert pipeline._load_token_cache()["access_token"] == "access" +def test_token_cache_plaintext_is_current_format_on_non_windows( + tmp_path, monkeypatch +) -> None: + monkeypatch.setattr(secure_token_cache.os, "name", "posix") + pipeline = _pipeline(tmp_path, monkeypatch) + cache_path = tmp_path / "signing_files" / ".token_cache.json" + cache_path.parent.mkdir() + cache_path.write_text( + json.dumps( + { + "creation_date": date.today().isoformat(), + "access_token": "access", + "refresh_token": "refresh", + "user_id": "user", + "jwt_token": "jwt", + } + ), + encoding="utf-8", + ) + rewrite = Mock() + monkeypatch.setattr(pipeline, "_write_token_cache", rewrite) + + assert pipeline._load_token_cache()["access_token"] == "access" + rewrite.assert_not_called() + + +def test_authenticate_reuses_token_without_browser_or_device( + tmp_path, monkeypatch +) -> None: + pipeline = _pipeline(tmp_path, monkeypatch) + cache = { + "creation_date": date.today().isoformat(), + "access_token": "access", + "refresh_token": "refresh", + "user_id": "user", + "jwt_token": "jwt", + } + monkeypatch.setattr(pipeline, "_load_token_cache", Mock(return_value=cache)) + init_client = Mock() + monkeypatch.setattr(pipeline, "_init_client_from_cache", init_client) + monkeypatch.setattr( + pipeline, + "_step_login", + Mock(side_effect=AssertionError("browser should not open")), + ) + monkeypatch.setattr( + pipeline, + "_step_check_device", + Mock(side_effect=AssertionError("device should not be queried")), + ) + + result = pipeline.authenticate() + + assert result == { + "authenticated": True, + "from_cache": True, + "creation_date": date.today().isoformat(), + } + init_client.assert_called_once_with(cache) + + def test_token_cache_bad_base64_falls_back_to_relogin(tmp_path, monkeypatch) -> None: pipeline = _pipeline(tmp_path, monkeypatch) cache_path = tmp_path / "signing_files" / ".token_cache.json" @@ -308,6 +378,8 @@ def test_pipeline_failure_logs_redacted_exception( assert result is False assert secret not in caplog.text assert "tempToken=" in caplog.text + assert secret not in pipeline.last_error + assert "tempToken=" in pipeline.last_error def test_token_cache_protect_failure_keeps_old_file(tmp_path, monkeypatch) -> None: @@ -359,6 +431,29 @@ def test_metadata_does_not_store_keystore_password(tmp_path, monkeypatch) -> Non assert "keystore_password" not in metadata +def test_metadata_for_another_device_is_ignored(tmp_path, monkeypatch) -> None: + pipeline = _pipeline(tmp_path, monkeypatch) + paths = {} + for suffix in ("p12", "cer", "p7b"): + path = tmp_path / "signing" / f"material.{suffix}" + path.write_bytes(b"placeholder") + paths[suffix] = str(path) + pipeline._udid = "B" * 64 + with open(pipeline._metadata_path, "w", encoding="utf-8") as metadata: + json.dump( + { + "creation_date": date.today().isoformat(), + "udid": "A" * 64, + "p12_path": paths["p12"], + "cer_path": paths["cer"], + "p7b_path": paths["p7b"], + }, + metadata, + ) + + assert pipeline._load_cached_metadata() is None + + def test_extract_permissions_filters_non_acl_entries(tmp_path, monkeypatch) -> None: pipeline = _pipeline(tmp_path, monkeypatch) allowed = next(iter(ACL_PERMISSION_WHITELIST)) @@ -468,6 +563,72 @@ def test_run_installs_already_signed_hap_directly(tmp_path, monkeypatch) -> None pipeline._step_sign_hap.assert_not_called() +def test_sign_only_keeps_signed_hap_without_installing(tmp_path, monkeypatch) -> None: + from hapsign import pipeline as pipeline_module + + hap = tmp_path / "signed.hap" + hap.write_bytes(b"placeholder") + values = [] + pipeline = SignPipeline( + hap_path=str(hap), + bundle_name="com.example.app", + work_dir=str(tmp_path / "signing"), + install_after_sign=False, + progress_callback=lambda value, label: values.append((value, label)), + ) + monkeypatch.setattr(pipeline_module, "is_hap_signed", lambda _path: True) + install = Mock(side_effect=AssertionError("sign must not install")) + monkeypatch.setattr( + pipeline_module, + "Installer", + lambda **_kwargs: SimpleNamespace( + get_udid=lambda: "A" * 64, + install=install, + close=Mock(), + ), + ) + + assert pipeline.run() is True + assert pipeline.signed_hap_path == str(hap) + assert values[-1] == (100, "签名完成") + install.assert_not_called() + + +def test_unsigned_sign_only_omits_install_step(tmp_path, monkeypatch) -> None: + from hapsign import pipeline as pipeline_module + + pipeline = SignPipeline( + hap_path=str(tmp_path / "unsigned.hap"), + bundle_name="com.example.app", + work_dir=str(tmp_path / "signing"), + install_after_sign=False, + ) + monkeypatch.setattr(pipeline_module, "is_hap_signed", lambda _path: False) + monkeypatch.setattr(pipeline, "_step_check_device", Mock()) + monkeypatch.setattr( + pipeline, + "_load_cached_metadata", + Mock( + return_value={ + "p12_path": "key.p12", + "cer_path": "cert.cer", + "p7b_path": "profile.p7b", + } + ), + ) + labels = [] + + def _capture_steps(steps): + labels.extend(label for label, _step in steps) + return True + + monkeypatch.setattr(pipeline, "_run_steps", _capture_steps) + + assert pipeline._run_pipeline() is True + assert labels == ["检测设备连接", "签名 hap"] + assert "安装 hap 到设备" not in labels + + def test_run_unsigned_hap_does_not_take_signed_shortcut(tmp_path, monkeypatch) -> None: from hapsign import pipeline as pipeline_module @@ -566,6 +727,7 @@ def test_pipeline_runtime_state_initialized(tmp_path, monkeypatch) -> None: assert pipeline._csr_path == "" assert pipeline._csr_content == "" assert pipeline._signed_hap_path == "" + assert pipeline.last_error == "" def test_force_refresh_signing_decision_does_not_leak(tmp_path, monkeypatch) -> None: diff --git a/tests/test_signing.py b/tests/test_signing.py index f18caca..1af34f1 100644 --- a/tests/test_signing.py +++ b/tests/test_signing.py @@ -62,6 +62,87 @@ def test_installer_extracts_udid(monkeypatch) -> None: assert installer.Installer().get_udid() == udid +def test_installer_targets_explicit_serial(monkeypatch) -> None: + udid = "A" * 64 + run = Mock(return_value=SimpleNamespace(returncode=0, stdout=udid, stderr="")) + monkeypatch.setattr(installer, "run_process", run) + monkeypatch.setattr(installer.Installer, "_ensure_server", lambda self: None) + + assert installer.Installer(serial="device-serial").get_udid() == udid + assert run.call_args.args[0][:3] == [ + installer.config.HDC_PATH, + "-t", + "device-serial", + ] + + +def test_installer_lists_agent_friendly_targets(monkeypatch) -> None: + result = SimpleNamespace( + returncode=0, + stdout=( + "127.0.0.1:5555\t\tTCP\tConnected\tlocalhost\n" + "5XQ0225613000233\t\tUSB\tConnected\tlocalhost\n" + "offline-device\t\tUSB\tOffline\tlocalhost\n" + ), + stderr="", + ) + run = Mock(return_value=result) + monkeypatch.setattr(installer, "run_process", run) + monkeypatch.setattr(installer.Installer, "_ensure_server", lambda self: None) + + targets = installer.Installer().list_targets(connected_only=True) + + assert [target["serial"] for target in targets] == [ + "127.0.0.1:5555", + "5XQ0225613000233", + ] + assert targets[0]["likely_emulator"] is True + assert targets[0]["physical_candidate"] is False + assert targets[1]["likely_emulator"] is False + assert targets[1]["physical_candidate"] is True + assert targets[1]["localhost"] is False + assert run.call_args.args[0] == [ + installer.config.HDC_PATH, + "list", + "targets", + "-v", + ] + + +def test_installer_inspects_bundle_on_explicit_serial(monkeypatch) -> None: + result = SimpleNamespace( + returncode=0, + stdout=( + '{"bundleName":"com.example.app","appProvisionType":"debug",' + '"versionName":"1.2.3"}' + ), + stderr="", + ) + run = Mock(return_value=result) + monkeypatch.setattr(installer, "run_process", run) + monkeypatch.setattr(installer.Installer, "_ensure_server", lambda self: None) + + bundle = installer.Installer(serial="device-serial").inspect_bundle( + "com.example.app" + ) + + assert bundle == { + "bundle_name": "com.example.app", + "provision_type": "debug", + "version_name": "1.2.3", + } + assert run.call_args.args[0] == [ + installer.config.HDC_PATH, + "-t", + "device-serial", + "shell", + "bm", + "dump", + "-n", + "com.example.app", + ] + + def test_installer_raises_on_fail_marker(monkeypatch) -> None: # rc=0 但输出含 [Fail] 状态行 → 判定失败 result = SimpleNamespace( @@ -119,6 +200,36 @@ def test_installer_raises_on_nonzero_exit(monkeypatch) -> None: installer.Installer().install("app.hap") +def test_installer_requires_success_text(monkeypatch) -> None: + result = SimpleNamespace(returncode=0, stdout="transfer finished", stderr="") + monkeypatch.setattr(installer, "run_process", Mock(return_value=result)) + monkeypatch.setattr(installer.Installer, "_ensure_server", lambda self: None) + + with pytest.raises(RuntimeError, match="hdc install"): + installer.Installer().install("app.hap") + + +def test_installer_uses_serial_and_replace(monkeypatch) -> None: + result = SimpleNamespace( + returncode=0, + stdout="install bundle successfully", + stderr="", + ) + run = Mock(return_value=result) + monkeypatch.setattr(installer, "run_process", run) + monkeypatch.setattr(installer.Installer, "_ensure_server", lambda self: None) + + assert installer.Installer(serial="device-serial").install("app.hap") is True + assert run.call_args.args[0] == [ + installer.config.HDC_PATH, + "-t", + "device-serial", + "install", + "-r", + "app.hap", + ] + + def test_installer_accepts_success_with_error_in_path(monkeypatch) -> None: # 路径/包名含 error/failed 但安装成功 → 不得误判 result = SimpleNamespace( From 4c7bedfd6668efb4dc6c9739650bb767cc6fb363 Mon Sep 17 00:00:00 2001 From: stefan Date: Fri, 28 Aug 2026 17:05:51 +0800 Subject: [PATCH 2/4] docs: add repository HapSign deploy skill --- .agents/skills/hapsign-hap-deploy/SKILL.md | 38 ++++++++++++++++++++++ 1 file changed, 38 insertions(+) create mode 100644 .agents/skills/hapsign-hap-deploy/SKILL.md diff --git a/.agents/skills/hapsign-hap-deploy/SKILL.md b/.agents/skills/hapsign-hap-deploy/SKILL.md new file mode 100644 index 0000000..f2aa42e --- /dev/null +++ b/.agents/skills/hapsign-hap-deploy/SKILL.md @@ -0,0 +1,38 @@ +--- +name: hapsign-hap-deploy +description: Authenticate, sign, install, or deploy unsigned and signed HarmonyOS HAPs on an explicitly selected connected device through the hapsign CLI. Use for HAP deployment; not for building HAPs, emulator-only workflows, or manual DevEco/HDC signing. +--- + +# HapSign HAP Deploy + +Use the installed `hapsign` CLI as the only implementation layer. It owns Huawei +authentication, token caching, device-bound Profiles, signing, HDC installation, +and post-install bundle checks. Do not recreate those steps with DevEco, raw HDC, +or skill-local scripts. + +## Commands + +Always request JSON and pass an explicit HDC serial: + +```bash +hapsign devices list --connected-only --json +hapsign auth status --json +hapsign auth --json +hapsign sign --hap --serial --json +hapsign install --hap --serial --json +hapsign deploy --hap --serial --json +``` + +- Honor a serial supplied by the user. Otherwise select the sole + `physical_candidate=true` target; ask the user if zero or multiple physical + candidates remain. Do not select `likely_emulator=true` unless requested. +- Use `sign` for signing only, `install` for an already signed HAP, and `deploy` + for end-to-end signing and installation. `deploy` also accepts signed HAPs. +- Check `auth status` before an operation that may need login. If no current cache + exists and the request does not already authorize account authentication, tell + the user that browser authorization is required before running `auth`. + +Treat success as exit code `0` plus JSON `ok=true`. Parse stdout as JSON and treat +stderr as diagnostic logs. On failure, report `error.type`, `error.message`, and +the exit code; do not fall back to manual signing or raw HDC installation. Never +print or copy tokens, passwords, signing keys, Profiles, or device UDIDs. From 3867f4c17377d51a4d351f3b3b50597cae144ef5 Mon Sep 17 00:00:00 2001 From: stefan Date: Fri, 28 Aug 2026 17:47:30 +0800 Subject: [PATCH 3/4] fix: redact device UDIDs and classify HDC failures --- CHANGELOG.md | 4 ++ hapsign/cli.py | 5 +- hapsign/diagnostics.py | 6 ++- tests/test_cli.py | 102 ++++++++++++++++++++++++++++++++++++++ tests/test_diagnostics.py | 14 ++++++ tests/test_pipeline.py | 7 ++- 6 files changed, 133 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 916c829..ea8eda0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -56,6 +56,8 @@ ### Fixed +- CLI 不再把缺少 HDC 可执行文件归类为输入错误；`devices`、`install` 等运行时 + HDC 失败现在返回 `operation_failed` 和退出码 1。 - HTTP 客户端正确发送 `User-Agent` / `Accept-Language` 请求头。 - Token 缓存缺少 `jwt_token` 时不再复用，避免后续刷新失败。 - 设备注册将业务层重复错误码视为成功，并保留 HTTP 错误信息中的兼容判定。 @@ -74,6 +76,8 @@ ### Security +- CLI、Pipeline 和诊断日志会脱敏异常文本中的完整 64 位设备 UDID；失败 JSON + stdout 不再泄露 `DeviceAPI.find_device_id()` 等异常携带的设备标识。 - 登录回调服务仅监听 loopback 地址。 - 日志默认不包含 token、完整请求体、CSRF code 或完整登录 URL；只有用户主动开启 “敏感诊断”且使用 DEBUG 级别时才记录完整网络载荷，密钥库密码始终排除。 diff --git a/hapsign/cli.py b/hapsign/cli.py index 3255fee..8b72f53 100644 --- a/hapsign/cli.py +++ b/hapsign/cli.py @@ -346,11 +346,12 @@ def _failure( *, error_type: str = "operation_failed", ) -> None: + safe_message = redact_sensitive_text(message) _emit( { "ok": False, "command": command, - "error": {"type": error_type, "message": message}, + "error": {"type": error_type, "message": safe_message}, }, json_output, ) @@ -563,7 +564,7 @@ def main(argv: Sequence[str] | None = None) -> int: except OperationCancelled: _failure(args.command, "操作已取消", args.json_output, error_type="cancelled") return EXIT_CANCELLED - except (ValueError, FileNotFoundError, zipfile.BadZipFile) as exc: + except (ValueError, zipfile.BadZipFile) as exc: message = str(redact_sensitive_text(exc)) _failure(args.command, message, args.json_output, error_type="invalid_input") return EXIT_USAGE diff --git a/hapsign/diagnostics.py b/hapsign/diagnostics.py index 58cdfb2..b93b0ac 100644 --- a/hapsign/diagnostics.py +++ b/hapsign/diagnostics.py @@ -28,6 +28,7 @@ r"(?P[^&,\s}'\"\]\)]+)" ) _BEARER_TOKEN_RE = re.compile(r"(?i)(?P\bBearer\s+)(?P[^\s,}'\"\]\)]+)") +_DEVICE_UDID_RE = re.compile(r"(?i)(? None: @@ -42,7 +43,7 @@ def sensitive_logging_enabled() -> bool: def redact_sensitive_text(value: object) -> str: - """脱敏异常文本中的令牌、密码和授权参数。 + """脱敏异常文本中的令牌、密码、授权参数和设备 UDID。 敏感诊断开关是显式选择，开启后保留原始文本以便定位协议问题；默认日志 必须避免把认证信息写入控制台、文件或 GUI 日志窗口。 @@ -62,10 +63,11 @@ def redact_sensitive_text(value: object) -> str: lambda match: f"{match.group('key')}{match.group('separator')}", text, ) - return _BEARER_TOKEN_RE.sub( + text = _BEARER_TOKEN_RE.sub( lambda match: f"{match.group('prefix')}", text, ) + return _DEVICE_UDID_RE.sub("", text) def configure_file_logging(settings: AppSettings) -> Path: diff --git a/tests/test_cli.py b/tests/test_cli.py index 7486287..ad795c7 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -118,6 +118,43 @@ def run(self): assert payload["error"]["message"] == ("检测设备连接: device unavailable") +def test_failure_json_redacts_pipeline_device_udid( + monkeypatch, tmp_path, capsys +) -> None: + hap_path = tmp_path / "example.hap" + _write_hap(hap_path) + device_udid = "A" * 64 + + class FakePipeline: + def __init__(self, **_kwargs): + self.last_error = f"Device not found in list: {device_udid}" + + def run(self): + return False + + monkeypatch.setattr(cli, "SignPipeline", FakePipeline) + monkeypatch.setattr(cli, "is_hap_signed", lambda _path: False) + + assert ( + cli.main( + [ + "sign", + "--hap", + str(hap_path), + "--serial", + "device-serial", + "--json", + ] + ) + == 1 + ) + + stdout = capsys.readouterr().out + assert device_udid not in stdout + payload = json.loads(stdout) + assert payload["error"]["message"] == ("Device not found in list: ") + + def test_sign_json_does_not_install(monkeypatch, tmp_path, capsys) -> None: hap_path = tmp_path / "unsigned.hap" signed_path = tmp_path / "unsigned_signed.hap" @@ -255,6 +292,28 @@ def list_targets(self, connected_only=False): assert payload["targets"] == targets +def test_devices_missing_hdc_is_operation_failure(monkeypatch, capsys) -> None: + class MissingHdcInstaller: + def __enter__(self): + return self + + def __exit__(self, *_args): + return None + + def list_targets(self, connected_only=False): + raise FileNotFoundError(2, "No such file or directory", "hdc") + + monkeypatch.setattr(cli, "Installer", MissingHdcInstaller) + + assert cli.main(["devices", "list", "--json"]) == 1 + + payload = json.loads(capsys.readouterr().out) + assert payload["ok"] is False + assert payload["command"] == "devices" + assert payload["error"]["type"] == "operation_failed" + assert "hdc" in payload["error"]["message"] + + def test_json_argument_error_has_stable_exit_code(capsys) -> None: assert cli.main(["sign", "--hap", "app.hap", "--json"]) == 2 @@ -340,6 +399,49 @@ def inspect_bundle(self, bundle_name): ] +def test_install_missing_hdc_is_operation_failure( + monkeypatch, tmp_path, capsys +) -> None: + hap_path = tmp_path / "signed.hap" + _write_hap(hap_path) + + class MissingHdcInstaller: + def __init__(self, serial=None): + pass + + def __enter__(self): + return self + + def __exit__(self, *_args): + return None + + def install(self, path): + raise FileNotFoundError(2, "No such file or directory", "hdc") + + monkeypatch.setattr(cli, "Installer", MissingHdcInstaller) + monkeypatch.setattr(cli, "is_hap_signed", lambda _path: True) + + assert ( + cli.main( + [ + "install", + "--hap", + str(hap_path), + "--serial", + "device-serial", + "--json", + ] + ) + == 1 + ) + + payload = json.loads(capsys.readouterr().out) + assert payload["ok"] is False + assert payload["command"] == "install" + assert payload["error"]["type"] == "operation_failed" + assert "hdc" in payload["error"]["message"] + + def test_install_rejects_unsigned_hap(monkeypatch, tmp_path, capsys) -> None: hap_path = tmp_path / "unsigned.hap" _write_hap(hap_path) diff --git a/tests/test_diagnostics.py b/tests/test_diagnostics.py index 3a4beed..62e31f5 100644 --- a/tests/test_diagnostics.py +++ b/tests/test_diagnostics.py @@ -64,3 +64,17 @@ def test_redact_sensitive_text_hides_tokens_and_bearer_headers() -> None: assert "tempToken=" in text assert "accessToken=" in text assert "Authorization: Bearer " in text + + +def test_redact_sensitive_text_hides_device_udid() -> None: + diagnostics.set_sensitive_logging(False) + udid = "A" * 64 + + text = diagnostics.redact_sensitive_text( + f"Device not found in list: {udid}; request_id={udid}0" + ) + + assert udid not in text.split(";", 1)[0] + assert "Device not found in list: " in text + # 不应截断更长的十六进制诊断标识。 + assert f"request_id={udid}0" in text diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index 55a8f3e..a50a96a 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -359,6 +359,7 @@ def test_pipeline_failure_logs_redacted_exception( set_sensitive_logging(False) pipeline = _pipeline(tmp_path, monkeypatch) secret = "secret-temp-token" + device_udid = "A" * 64 caplog.set_level(logging.DEBUG, logger="hapsign.pipeline") result = pipeline._run_steps( @@ -368,7 +369,8 @@ def test_pipeline_failure_logs_redacted_exception( Mock( side_effect=RuntimeError( "request failed: " - f"/authrouter/auth/api/temptoken/check?tempToken={secret}" + f"/authrouter/auth/api/temptoken/check?tempToken={secret}; " + f"Device not found in list: {device_udid}" ) ), ) @@ -380,6 +382,9 @@ def test_pipeline_failure_logs_redacted_exception( assert "tempToken=" in caplog.text assert secret not in pipeline.last_error assert "tempToken=" in pipeline.last_error + assert device_udid not in caplog.text + assert device_udid not in pipeline.last_error + assert "Device not found in list: " in pipeline.last_error def test_token_cache_protect_failure_keeps_old_file(tmp_path, monkeypatch) -> None: From 3893184a7459f7d6be0e615ebda61572b9ff9122 Mon Sep 17 00:00:00 2001 From: guantw Date: Fri, 28 Aug 2026 19:24:11 +0800 Subject: [PATCH 4/4] fix: harden agent CLI failure handling --- CHANGELOG.md | 9 ++++++ hapsign/cli.py | 32 ++++++++++-------- hapsign/pipeline.py | 18 ++++++----- hapsign/signing/installer.py | 13 ++++++-- tests/test_cli.py | 63 ++++++++++++++++++++++++++++++++---- tests/test_pipeline.py | 16 +++++++-- tests/test_signing.py | 22 +++++++++++++ 7 files changed, 138 insertions(+), 35 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ea8eda0..ba4f829 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -54,8 +54,17 @@ 哈希、许可材料及 libusb 对应源码。DevEco 工具链只保留为显式排障回退。 - 便携构建在 ZIP 旁自动生成标准 `.sha256` 校验文件，降低发布时手工抄录哈希出错的风险。 +### Changed + +- CLI 现在必须显式使用 `auth`、`devices`、`sign`、`install` 或 `deploy` + 子命令，并为设备相关命令传入非空 `--serial`；旧的 `hapsign --hap ...` + 调用方式不再兼容。 + ### Fixed +- Agent CLI 会拒绝空白 HDC serial，避免退回隐式设备选择；`auth` 仅在 Token + 缓存成功落盘后返回成功；`devices list` 不再把退出码为 0 的 HDC `[Fail]` + 输出误报为空设备列表。 - CLI 不再把缺少 HDC 可执行文件归类为输入错误；`devices`、`install` 等运行时 HDC 失败现在返回 `operation_failed` 和退出码 1。 - HTTP 客户端正确发送 `User-Agent` / `Accept-Language` 请求头。 diff --git a/hapsign/cli.py b/hapsign/cli.py index 8b72f53..2c238c4 100644 --- a/hapsign/cli.py +++ b/hapsign/cli.py @@ -68,6 +68,14 @@ def _add_state_option(parser: argparse.ArgumentParser) -> None: ) +def _nonempty_serial(value: str) -> str: + """规范化显式 HDC serial，禁止空值退回隐式目标选择。""" + serial = value.strip() + if not serial: + raise argparse.ArgumentTypeError("--serial 不能为空") + return serial + + def _add_hap_identity_options(parser: argparse.ArgumentParser) -> None: parser.add_argument("--hap", required=True, help="HAP 文件的绝对或相对路径") parser.add_argument( @@ -82,6 +90,7 @@ def _add_signing_options(parser: argparse.ArgumentParser) -> None: parser.add_argument( "--serial", required=True, + type=_nonempty_serial, help="hdc list targets 返回的目标序列号；Profile 将绑定该设备", ) parser.add_argument("--country", default="CN", help="华为账号国家码；默认 CN") @@ -143,9 +152,6 @@ def build_parser() -> argparse.ArgumentParser: hapsign install --hap app-signed.hap --serial --json hapsign deploy --hap app-unsigned.hap --serial --json -兼容旧调用: - hapsign --hap app.hap --serial 等价于 hapsign deploy ... - 退出码: 0 成功 1 运行失败（认证、签名、HDC 或安装失败） @@ -251,7 +257,12 @@ def build_parser() -> argparse.ArgumentParser: """, ) _add_hap_identity_options(install) - install.add_argument("--serial", required=True, help="目标 HDC 序列号") + install.add_argument( + "--serial", + required=True, + type=_nonempty_serial, + help="目标 HDC 序列号", + ) _add_output_options(install) deploy = subparsers.add_parser( @@ -272,14 +283,6 @@ def build_parser() -> argparse.ArgumentParser: return parser -def _normalize_legacy_args(argv: Sequence[str]) -> list[str]: - """把旧 ``hapsign --hap ...`` 调用转换成 ``deploy`` 子命令。""" - normalized = list(argv) - if "--hap" in normalized and not any(item in COMMANDS for item in normalized[:1]): - normalized.insert(0, "deploy") - return normalized - - def _command_from_args(argv: Sequence[str]) -> str: return next((item for item in argv if item in COMMANDS), "unknown") @@ -411,6 +414,8 @@ def _run_auth(args: argparse.Namespace) -> int: result = pipeline.authenticate(force_refresh=args.refresh) status = pipeline.auth_status() + if not status.get("authenticated"): + raise RuntimeError("认证成功，但 Token 缓存未能持久化") result.update( { "ok": True, @@ -541,8 +546,7 @@ def main(argv: Sequence[str] | None = None) -> int: """运行 CLI 并返回稳定退出码；JSON 模式永不输出凭据。""" raw_args = list(sys.argv[1:] if argv is None else argv) parser = build_parser() - normalized_args = _normalize_legacy_args(raw_args) - parsed = _parse_args(parser, normalized_args) + parsed = _parse_args(parser, raw_args) if isinstance(parsed, int): return parsed args = parsed diff --git a/hapsign/pipeline.py b/hapsign/pipeline.py index 26354fe..fc68699 100644 --- a/hapsign/pipeline.py +++ b/hapsign/pipeline.py @@ -219,10 +219,10 @@ def _load_token_cache(self) -> dict | None: self._write_token_cache(cache) return cache - def _save_token_cache(self) -> None: + def _save_token_cache(self) -> bool: """保存当前 token 信息到缓存文件。""" if self._token_info is None: - return + return False cache = { "creation_date": date.today().isoformat(), "access_token": self._token_info.access_token, @@ -233,9 +233,9 @@ def _save_token_cache(self) -> None: "real_name": self._token_info.real_name, "jwt_token": self._token_info.jwt_token, } - self._write_token_cache(cache) + return self._write_token_cache(cache) - def _write_token_cache(self, cache: dict) -> None: + def _write_token_cache(self, cache: dict) -> bool: """把 token 缓存写入磁盘（DPAPI 加密；原子替换，失败保留原缓存）。""" os.makedirs(self.state_dir, exist_ok=True) tmp_path = self._token_cache_path + ".tmp" @@ -262,8 +262,9 @@ def _write_token_cache(self, cache: dict) -> None: "保存 token 缓存失败，保留原缓存: %s", redact_sensitive_text(exc), ) - return + return False logger.info("[cache] token 缓存已保存: %s", self._token_cache_path) + return True def _init_client_from_cache(self, cache: dict) -> None: """从缓存数据初始化 API 客户端。""" @@ -528,7 +529,7 @@ def _step_login(self) -> None: ) self._temp_token = login.login(self.country) - def _step_exchange_token(self) -> None: + def _step_exchange_token(self) -> bool: """tempToken -> jwtToken -> accessToken，初始化 API 客户端，保存 token 缓存。""" jwt_token = self._token_exchange.exchange_temp_token( self._temp_token, self.country @@ -548,7 +549,7 @@ def _step_exchange_token(self) -> None: self._capability_api = CapabilityAPI(self._client) # 保存 token 缓存，供同一天内复用 - self._save_token_cache() + return self._save_token_cache() def authenticate(self, force_refresh: bool = False) -> dict[str, object]: """确保账号认证缓存可用，不执行设备检测、签名或安装。""" @@ -564,7 +565,8 @@ def authenticate(self, force_refresh: bool = False) -> dict[str, object]: } self._step_login() - self._step_exchange_token() + if not self._step_exchange_token(): + raise RuntimeError("认证成功，但 Token 缓存写入失败") return { "authenticated": True, "from_cache": False, diff --git a/hapsign/signing/installer.py b/hapsign/signing/installer.py index b861985..de5479f 100644 --- a/hapsign/signing/installer.py +++ b/hapsign/signing/installer.py @@ -31,6 +31,10 @@ _FATAL_INSTALL_MARKERS = re.compile( r"(?:\[fail\]|install_failed_|^error:)", re.IGNORECASE | re.MULTILINE ) +_FATAL_HDC_OUTPUT = re.compile( + r"^\s*(?:\[fail\]|error:|connect server failed\b)", + re.IGNORECASE | re.MULTILINE, +) def _listener_pid() -> int | None: @@ -172,7 +176,7 @@ def __init__( def _device_command(self, *args: str) -> list[str]: """构建设备命令；指定 serial 时显式选择目标，避免 HDC 隐式歧义。""" command = [self._hdc] - if self.serial: + if self.serial is not None: command.extend(["-t", self.serial]) command.extend(args) return command @@ -343,8 +347,11 @@ def list_targets(self, connected_only: bool = False) -> list[dict[str, object]]: timeout=15, cancel_event=self.cancel_event, ) - if result.returncode != 0: - output = (result.stderr or result.stdout or "").strip() + output = "\n".join( + part for part in (result.stdout or "", result.stderr or "") if part + ) + if result.returncode != 0 or _FATAL_HDC_OUTPUT.search(output): + output = output.strip() raise RuntimeError(f"hdc list targets 失败: {output}") targets: list[dict[str, object]] = [] diff --git a/tests/test_cli.py b/tests/test_cli.py index ad795c7..612a52d 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -38,12 +38,6 @@ def test_detect_bundle_name_requires_bundle_name(tmp_path) -> None: cli.detect_bundle_name(str(hap_path)) -def test_legacy_hap_invocation_maps_to_deploy() -> None: - assert cli._normalize_legacy_args( - ["--hap", "app.hap", "--serial", "device-serial"] - ) == ["deploy", "--hap", "app.hap", "--serial", "device-serial"] - - def test_parser_uses_home_state_dir_by_default(monkeypatch, tmp_path) -> None: expected = str(tmp_path / ".hapsign") monkeypatch.setattr(cli, "default_state_dir", lambda: expected) @@ -245,7 +239,10 @@ def authenticate(self, force_refresh=False): } def auth_status(self): - return {"cache_path": str(cache_path)} + return { + "authenticated": True, + "cache_path": str(cache_path), + } monkeypatch.setattr(cli, "SignPipeline", FakePipeline) @@ -256,6 +253,38 @@ def auth_status(self): assert payload["from_cache"] is True +def test_auth_login_fails_when_cache_was_not_persisted( + monkeypatch, tmp_path, capsys +) -> None: + cache_path = tmp_path / ".token_cache.json" + + class FakePipeline: + def __init__(self, **_kwargs): + pass + + def authenticate(self, force_refresh=False): + return { + "authenticated": True, + "from_cache": False, + "creation_date": "2026-08-28", + } + + def auth_status(self): + return { + "authenticated": False, + "cache_path": str(cache_path), + } + + monkeypatch.setattr(cli, "SignPipeline", FakePipeline) + + assert cli.main(["auth", "--json"]) == 1 + + payload = json.loads(capsys.readouterr().out) + assert payload["ok"] is False + assert payload["error"]["type"] == "operation_failed" + assert "Token 缓存" in payload["error"]["message"] + + def test_devices_list_json(monkeypatch, capsys) -> None: targets = [ { @@ -324,6 +353,26 @@ def test_json_argument_error_has_stable_exit_code(capsys) -> None: assert "--serial" in payload["error"]["message"] +def test_legacy_hap_invocation_is_rejected(capsys) -> None: + assert cli.main(["--hap", "app.hap", "--serial", "device-serial", "--json"]) == 2 + + payload = json.loads(capsys.readouterr().out) + assert payload["ok"] is False + assert payload["command"] == "unknown" + assert payload["error"]["type"] == "invalid_arguments" + + +@pytest.mark.parametrize("command", ["sign", "install", "deploy"]) +def test_empty_serial_is_rejected_as_invalid_arguments(command, capsys) -> None: + assert cli.main([command, "--hap", "app.hap", "--serial", " ", "--json"]) == 2 + + payload = json.loads(capsys.readouterr().out) + assert payload["ok"] is False + assert payload["command"] == command + assert payload["error"]["type"] == "invalid_arguments" + assert "--serial" in payload["error"]["message"] + + def test_missing_hap_json_is_invalid_input(capsys, tmp_path) -> None: missing = tmp_path / "missing.hap" diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index a50a96a..d88ad2a 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -70,7 +70,7 @@ def test_token_cache_round_trip(tmp_path, monkeypatch) -> None: jwt_token="jwt", ) - pipeline._save_token_cache() + assert pipeline._save_token_cache() is True assert pipeline._load_token_cache()["access_token"] == "access" assert not (tmp_path / "signing_files" / ".token_cache.json.tmp").exists() @@ -92,7 +92,7 @@ def _broken_dumps(*args, **kwargs): monkeypatch.setattr("hapsign.pipeline.json.dumps", _broken_dumps) - pipeline._save_token_cache() + assert pipeline._save_token_cache() is False assert cache_path.read_text(encoding="utf-8") == '{"old": true}' assert not (tmp_path / "signing_files" / ".token_cache.json.tmp").exists() @@ -114,7 +114,7 @@ def _broken_replace(src, dst): monkeypatch.setattr("hapsign.pipeline.os.replace", _broken_replace) - pipeline._save_token_cache() + assert pipeline._save_token_cache() is False assert cache_path.read_text(encoding="utf-8") == '{"old": true}' assert not (tmp_path / "signing_files" / ".token_cache.json.tmp").exists() @@ -309,6 +309,16 @@ def test_authenticate_reuses_token_without_browser_or_device( init_client.assert_called_once_with(cache) +def test_authenticate_fails_when_token_cache_write_fails(tmp_path, monkeypatch) -> None: + pipeline = _pipeline(tmp_path, monkeypatch) + monkeypatch.setattr(pipeline, "_load_token_cache", Mock(return_value=None)) + monkeypatch.setattr(pipeline, "_step_login", Mock()) + monkeypatch.setattr(pipeline, "_step_exchange_token", Mock(return_value=False)) + + with pytest.raises(RuntimeError, match="Token 缓存写入失败"): + pipeline.authenticate() + + def test_token_cache_bad_base64_falls_back_to_relogin(tmp_path, monkeypatch) -> None: pipeline = _pipeline(tmp_path, monkeypatch) cache_path = tmp_path / "signing_files" / ".token_cache.json" diff --git a/tests/test_signing.py b/tests/test_signing.py index 1af34f1..de7f6b8 100644 --- a/tests/test_signing.py +++ b/tests/test_signing.py @@ -76,6 +76,15 @@ def test_installer_targets_explicit_serial(monkeypatch) -> None: ] +def test_installer_does_not_treat_explicit_empty_serial_as_implicit() -> None: + assert installer.Installer(serial="")._device_command("shell") == [ + installer.config.HDC_PATH, + "-t", + "", + "shell", + ] + + def test_installer_lists_agent_friendly_targets(monkeypatch) -> None: result = SimpleNamespace( returncode=0, @@ -109,6 +118,19 @@ def test_installer_lists_agent_friendly_targets(monkeypatch) -> None: ] +def test_installer_rejects_hdc_failure_marker_with_zero_exit(monkeypatch) -> None: + result = SimpleNamespace( + returncode=0, + stdout="[Fail]Connect server failed.\n", + stderr="", + ) + monkeypatch.setattr(installer, "run_process", Mock(return_value=result)) + monkeypatch.setattr(installer.Installer, "_ensure_server", lambda self: None) + + with pytest.raises(RuntimeError, match="hdc list targets"): + installer.Installer().list_targets() + + def test_installer_inspects_bundle_on_explicit_serial(monkeypatch) -> None: result = SimpleNamespace( returncode=0,