From 46f680345c65f5ae5458aa8b1a86e28d0c3d92f3 Mon Sep 17 00:00:00 2001 From: guantw Date: Tue, 1 Sep 2026 11:47:38 +0800 Subject: [PATCH] fix: make Windows tool output decoding explicit Parse localized netstat output as bytes and decode all HDC output as UTF-8. Report invalid captured output explicitly and avoid decoding localized taskkill output. --- CHANGELOG.md | 2 + hapsign/signing/installer.py | 25 ++++++-- hapsign/subprocess_utils.py | 84 +++++++++++++++++++++--- tests/test_signing.py | 36 ++++++++--- tests/test_subprocess_utils.py | 113 +++++++++++++++++++++++++++++++++ 5 files changed, 235 insertions(+), 25 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5b41534..455a93b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -93,6 +93,8 @@ ### Fixed +- Windows 设备检测改为从本地化 `netstat` 原始字节解析监听端口，HDC 输出固定按 + UTF-8 解码，避免 `PYTHONUTF8=1` 或系统 GBK 文本模式导致设备检测及安装后校验失败。 - Agent CLI 会拒绝空白 HDC serial，避免退回隐式设备选择；`auth` 仅在 Token 缓存成功落盘后返回成功；`devices list` 不再把退出码为 0 的 HDC `[Fail]` 输出误报为空设备列表。 diff --git a/hapsign/signing/installer.py b/hapsign/signing/installer.py index b9679be..5282ace 100644 --- a/hapsign/signing/installer.py +++ b/hapsign/signing/installer.py @@ -19,6 +19,8 @@ _HDC_SERVER_HOST = "127.0.0.1" _HDC_SERVER_PORT = 8710 +_HDC_OUTPUT_ENCODING = "utf-8" +_HDC_OUTPUT_ERRORS = "replace" # hdc start 调用时刻与监听进程创建时刻比较时允许的时钟偏差（秒） _CLOCK_SKEW_SECONDS = 2.0 @@ -47,20 +49,19 @@ def _listener_pid() -> int | None: def _listener_pid_windows() -> int | None: - """解析 netstat 输出，取本地监听 8710 的 PID。""" + """从 netstat 原始字节中解析本地监听 8710 的 PID。""" try: result = subprocess.run( ["netstat", "-ano", "-p", "TCP"], capture_output=True, - text=True, timeout=10, **no_window_kwargs(), ) except (OSError, subprocess.TimeoutExpired): return None # 监听套接字的远端地址恒为 0.0.0.0:0，状态列文本在不同语言环境可能不同 - pattern = re.compile(r"TCP\s+127\.0\.0\.1:8710\s+0\.0\.0\.0:0\s+\S+\s+(\d+)\s*$") - for line in result.stdout.splitlines(): + pattern = re.compile(rb"TCP\s+127\.0\.0\.1:8710\s+0\.0\.0\.0:0\s+\S+\s+(\d+)\s*$") + for line in (result.stdout or b"").splitlines(): match = pattern.search(line) if match: return int(match.group(1)) @@ -282,6 +283,8 @@ def _ensure_server(self) -> None: [self._hdc, "start"], capture_output=True, text=True, + encoding=_HDC_OUTPUT_ENCODING, + errors=_HDC_OUTPUT_ERRORS, timeout=10, **no_window_kwargs(), ) @@ -346,6 +349,8 @@ def close(self) -> None: [self._hdc, "kill"], capture_output=True, text=True, + encoding=_HDC_OUTPUT_ENCODING, + errors=_HDC_OUTPUT_ERRORS, timeout=5, **no_window_kwargs(), ) @@ -393,6 +398,8 @@ def get_udid(self) -> str: cmd, capture_output=True, text=True, + encoding=_HDC_OUTPUT_ENCODING, + errors=_HDC_OUTPUT_ERRORS, timeout=15, cancel_event=self.cancel_event, ) @@ -400,7 +407,7 @@ def get_udid(self) -> str: continue if result.returncode == 0: # 从输出中提取 64 位十六进制 UDID - match = re.search(r"\b([0-9A-Fa-f]{64})\b", result.stdout) + match = re.search(r"\b([0-9A-Fa-f]{64})\b", result.stdout or "") if match: return match.group(1) raise RuntimeError( @@ -414,6 +421,8 @@ def list_targets(self, connected_only: bool = False) -> list[dict[str, object]]: [self._hdc, "list", "targets", "-v"], capture_output=True, text=True, + encoding=_HDC_OUTPUT_ENCODING, + errors=_HDC_OUTPUT_ERRORS, timeout=15, cancel_event=self.cancel_event, ) @@ -425,7 +434,7 @@ def list_targets(self, connected_only: bool = False) -> list[dict[str, object]]: raise RuntimeError(f"hdc list targets 失败: {output}") targets: list[dict[str, object]] = [] - for raw_line in result.stdout.splitlines(): + for raw_line in (result.stdout or "").splitlines(): parts = raw_line.strip().split() if not parts or parts[0].startswith("["): continue @@ -460,6 +469,8 @@ def inspect_bundle(self, bundle_name: str) -> dict[str, str] | None: self._device_command("shell", "bm", "dump", "-n", bundle_name), capture_output=True, text=True, + encoding=_HDC_OUTPUT_ENCODING, + errors=_HDC_OUTPUT_ERRORS, timeout=20, cancel_event=self.cancel_event, ) @@ -500,6 +511,8 @@ def install(self, hap_path: str) -> bool: cmd, capture_output=True, text=True, + encoding=_HDC_OUTPUT_ENCODING, + errors=_HDC_OUTPUT_ERRORS, timeout=60, cancel_event=self.cancel_event, ) diff --git a/hapsign/subprocess_utils.py b/hapsign/subprocess_utils.py index 4f686d0..f789d03 100644 --- a/hapsign/subprocess_utils.py +++ b/hapsign/subprocess_utils.py @@ -15,6 +15,52 @@ _CREATE_NO_WINDOW = getattr(subprocess, "CREATE_NO_WINDOW", 0x08000000) +class ProcessOutputError(RuntimeError): + """外部进程的已捕获输出无法按调用方声明的格式提供。""" + + +def _output_encoding(text: bool, options: dict[str, Any]) -> str: + encoding = options.get("encoding") + if encoding is not None: + return str(encoding) + if text or options.get("errors") is not None: + return "Python default" + return "bytes" + + +def _process_output_error( + command: list[str], + text: bool, + options: dict[str, Any], + detail: str, +) -> ProcessOutputError: + # 只包含可执行文件名，避免把密码等敏感命令参数带入错误信息。 + executable = command[0] if command else "" + encoding = _output_encoding(text, options) + return ProcessOutputError( + f"外部命令 {executable!r} 的输出不可用 (encoding={encoding}): {detail}" + ) + + +def _validate_captured_output( + result: subprocess.CompletedProcess, + command: list[str], + *, + capture_output: bool, + text: bool, + options: dict[str, Any], +) -> subprocess.CompletedProcess: + """捕获开启时保证 stdout/stderr 存在，否则给出明确的边界错误。""" + if capture_output and (result.stdout is None or result.stderr is None): + raise _process_output_error( + command, + text, + options, + "capture_output=True 但 stdout/stderr 为 None", + ) + return result + + def no_window_kwargs() -> dict[str, int]: """Windows 下禁止控制台工具创建一闪而过的命令行窗口。""" if platform.system() == "Windows": @@ -150,8 +196,8 @@ def _terminate_windows_tree(pid: int) -> None: try: subprocess.run( ["taskkill", "/F", "/T", "/PID", str(pid)], - capture_output=True, - text=True, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, **no_window_kwargs(), ) except (OSError, ValueError): @@ -201,16 +247,27 @@ def run_process( """运行外部命令；有取消信号或超时要求时，终止子进程及整棵进程树。 仅当既无取消信号也无超时时走 subprocess.run 快路径；否则使用 Popen + - Job Object / 进程组，确保取消或超时能终止整棵进程树。 + Job Object / 进程组，确保取消或超时能终止整棵进程树。调用方要求捕获输出时， + stdout/stderr 必须保持稳定；解码失败或异常的 None 输出会转换为明确异常。 """ options = no_window_kwargs() options.update(kwargs) if cancel_event is None and timeout is None: - return subprocess.run( + try: + result = subprocess.run( + command, + capture_output=capture_output, + text=text, + **options, + ) + except UnicodeError as exc: + raise _process_output_error(command, text, options, str(exc)) from exc + return _validate_captured_output( + result, command, capture_output=capture_output, text=text, - **options, + options=options, ) raise_if_cancelled(cancel_event) @@ -248,11 +305,20 @@ def run_process( stdout, stderr = process.communicate(timeout=wait_timeout) except subprocess.TimeoutExpired: continue - return subprocess.CompletedProcess( + except UnicodeError as exc: + _stop_process(process, job) + raise _process_output_error(command, text, options, str(exc)) from exc + return _validate_captured_output( + subprocess.CompletedProcess( + command, + process.returncode, + stdout, + stderr, + ), command, - process.returncode, - stdout, - stderr, + capture_output=capture_output, + text=text, + options=options, ) finally: if job is not None: diff --git a/tests/test_signing.py b/tests/test_signing.py index ce8fd20..744b138 100644 --- a/tests/test_signing.py +++ b/tests/test_signing.py @@ -9,6 +9,12 @@ from hapsign.signing import hap_signer, installer, keytool_util +def _assert_hdc_utf8_call(call) -> None: + assert call.kwargs["text"] is True + assert call.kwargs["encoding"] == "utf-8" + assert call.kwargs["errors"] == "replace" + + def test_hap_signer_builds_subprocess_command(monkeypatch) -> None: run = Mock(return_value=SimpleNamespace(returncode=0, stdout="", stderr="")) monkeypatch.setattr(hap_signer, "run_process", run) @@ -60,6 +66,7 @@ def test_installer_extracts_udid(monkeypatch) -> None: monkeypatch.setattr(installer.Installer, "_ensure_server", lambda self: None) assert installer.Installer().get_udid() == udid + _assert_hdc_utf8_call(run.call_args) def test_installer_targets_explicit_serial(monkeypatch) -> None: @@ -74,6 +81,7 @@ def test_installer_targets_explicit_serial(monkeypatch) -> None: "-t", "device-serial", ] + _assert_hdc_utf8_call(run.call_args) def test_installer_does_not_treat_explicit_empty_serial_as_implicit() -> None: @@ -116,6 +124,7 @@ def test_installer_lists_agent_friendly_targets(monkeypatch) -> None: "targets", "-v", ] + _assert_hdc_utf8_call(run.call_args) def test_installer_rejects_hdc_failure_marker_with_zero_exit(monkeypatch) -> None: @@ -136,7 +145,7 @@ def test_installer_inspects_bundle_on_explicit_serial(monkeypatch) -> None: returncode=0, stdout=( '{"bundleName":"com.example.app","appProvisionType":"debug",' - '"versionName":"1.2.3"}' + '"versionName":"测试版 1.2.3"}' ), stderr="", ) @@ -151,7 +160,7 @@ def test_installer_inspects_bundle_on_explicit_serial(monkeypatch) -> None: assert bundle == { "bundle_name": "com.example.app", "provision_type": "debug", - "version_name": "1.2.3", + "version_name": "测试版 1.2.3", } assert run.call_args.args[0] == [ installer.config.HDC_PATH, @@ -163,6 +172,7 @@ def test_installer_inspects_bundle_on_explicit_serial(monkeypatch) -> None: "-n", "com.example.app", ] + _assert_hdc_utf8_call(run.call_args) def test_installer_raises_on_fail_marker(monkeypatch) -> None: @@ -250,6 +260,7 @@ def test_installer_uses_serial_and_replace(monkeypatch) -> None: "-r", "app.hap", ] + _assert_hdc_utf8_call(run.call_args) def test_installer_accepts_success_with_error_in_path(monkeypatch) -> None: @@ -282,6 +293,8 @@ def test_installer_closes_server_started_by_current_task(monkeypatch) -> None: commands = [call.args[0] for call in run.call_args_list] assert commands == [[hdc._hdc, "start"], [hdc._hdc, "kill"]] + for call in run.call_args_list: + _assert_hdc_utf8_call(call) def test_installer_preserves_preexisting_server(monkeypatch) -> None: @@ -399,29 +412,32 @@ def test_installer_polls_until_listener_appears(monkeypatch) -> None: def test_listener_pid_parses_netstat_output(monkeypatch) -> None: + # 本地化表头包含非法 UTF-8 字节；监听行本身只依赖 ASCII 字段。 output = ( - " TCP 127.0.0.1:8710 0.0.0.0:0 LISTENING 47024\n" - " TCP 127.0.0.1:8710 127.0.0.1:65140 TIME_WAIT 0\n" - " TCP 0.0.0.0:135 0.0.0.0:0 LISTENING 1234\n" + b"\xbb\xee\xb6\xaf\xd0\xad\xd2\xe9 \xb1\xbe\xb5\xd8\xb5\xd8\xd6\xb7\r\n" + b" TCP 127.0.0.1:8710 0.0.0.0:0 LISTENING 47024\r\n" + b" TCP 127.0.0.1:8710 127.0.0.1:65140 TIME_WAIT 0\r\n" + b" TCP 0.0.0.0:135 0.0.0.0:0 LISTENING 1234\r\n" ) monkeypatch.setattr(installer.os, "name", "nt") + run = Mock(return_value=SimpleNamespace(returncode=0, stdout=output, stderr=b"")) monkeypatch.setattr( installer.subprocess, "run", - Mock(return_value=SimpleNamespace(returncode=0, stdout=output, stderr="")), + run, ) assert installer._listener_pid() == 47024 + assert "text" not in run.call_args.kwargs + assert "encoding" not in run.call_args.kwargs def test_listener_pid_returns_none_when_port_free(monkeypatch) -> None: - output = ( - " TCP 0.0.0.0:135 0.0.0.0:0 LISTENING 1234\n" - ) + output = b" TCP 0.0.0.0:135 0.0.0.0:0 LISTENING 1234\r\n" monkeypatch.setattr(installer.os, "name", "nt") monkeypatch.setattr( installer.subprocess, "run", - Mock(return_value=SimpleNamespace(returncode=0, stdout=output, stderr="")), + Mock(return_value=SimpleNamespace(returncode=0, stdout=output, stderr=b"")), ) assert installer._listener_pid() is None diff --git a/tests/test_subprocess_utils.py b/tests/test_subprocess_utils.py index 71c1385..1576f59 100644 --- a/tests/test_subprocess_utils.py +++ b/tests/test_subprocess_utils.py @@ -40,6 +40,106 @@ def test_cancelled_command_does_not_start_process(monkeypatch) -> None: popen.assert_not_called() +def test_captured_text_output_none_raises_explicit_error(monkeypatch) -> None: + monkeypatch.setattr(subprocess_utils, "no_window_kwargs", lambda: {}) + monkeypatch.setattr( + subprocess_utils.subprocess, + "run", + Mock( + return_value=subprocess.CompletedProcess( + ["tool", "--secret", "not-in-error"], + 0, + None, + None, + ) + ), + ) + + with pytest.raises( + subprocess_utils.ProcessOutputError, + match=r"tool.*encoding=utf-8.*stdout/stderr.*None", + ) as caught: + subprocess_utils.run_process( + ["tool", "--secret", "not-in-error"], + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + ) + + assert "not-in-error" not in str(caught.value) + + +def test_captured_text_output_none_from_popen_raises_explicit_error( + monkeypatch, +) -> None: + process = Mock(returncode=0) + process.communicate.return_value = (None, None) + monkeypatch.setattr(subprocess_utils, "no_window_kwargs", lambda: {}) + monkeypatch.setattr(subprocess_utils, "_popen_process_tree_options", lambda: {}) + monkeypatch.setattr(subprocess_utils, "_create_job_object", lambda _process: None) + monkeypatch.setattr( + subprocess_utils.subprocess, "Popen", Mock(return_value=process) + ) + + with pytest.raises( + subprocess_utils.ProcessOutputError, + match=r"tool.*encoding=utf-8.*stdout/stderr.*None", + ): + subprocess_utils.run_process( + ["tool"], + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + timeout=1, + ) + + +def test_text_decode_failure_names_executable_and_encoding(monkeypatch) -> None: + monkeypatch.setattr(subprocess_utils, "no_window_kwargs", lambda: {}) + monkeypatch.setattr( + subprocess_utils.subprocess, + "run", + Mock( + side_effect=UnicodeDecodeError( + "utf-8", + b"\xff", + 0, + 1, + "invalid start byte", + ) + ), + ) + + with pytest.raises( + subprocess_utils.ProcessOutputError, + match=r"tool.*encoding=utf-8.*invalid start byte", + ): + subprocess_utils.run_process( + ["tool"], + capture_output=True, + text=True, + encoding="utf-8", + ) + + +def test_explicit_utf8_decodes_external_tool_output() -> None: + code = "import sys; sys.stdout.buffer.write('安装成功'.encode('utf-8'))" + + result = subprocess_utils.run_process( + [sys.executable, "-c", code], + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + timeout=5, + ) + + assert result.stdout == "安装成功" + assert result.stderr == "" + + def _spawning_child_code(out_file: str) -> str: """生成子进程代码：拉起一个挂起 300s 的孙进程，并把其 PID 写入文件。 @@ -182,3 +282,16 @@ def test_stop_process_falls_back_to_windows_tree_kill(monkeypatch) -> None: assert tree_killed == [process.pid] process.wait.assert_called_once_with(timeout=2) + + +def test_taskkill_output_is_discarded_without_text_decoding(monkeypatch) -> None: + run = Mock() + monkeypatch.setattr(subprocess_utils, "no_window_kwargs", lambda: {}) + monkeypatch.setattr(subprocess_utils.subprocess, "run", run) + + subprocess_utils._terminate_windows_tree(1234) + + assert run.call_args.args[0] == ["taskkill", "/F", "/T", "/PID", "1234"] + assert run.call_args.kwargs["stdout"] is subprocess.DEVNULL + assert run.call_args.kwargs["stderr"] is subprocess.DEVNULL + assert "text" not in run.call_args.kwargs