fix: secure Hibor input on replacement emulator

Cai

2026-08-09 892908574a00ccca683bd7483949e71dd632e27c

fix: secure Hibor input on replacement emulator

 dev-doc/ana-doc/开发方案/IMPLEMENTATION-ANA-HIBOR-NEW-EMULATOR-INPUT-COMPAT-V001.md

New file
@@ -0,0 +1,24 @@
# 慧博新模拟器输入兼容与安全修复实现记录 V001

- 事项：`DEV-ANA-HIBOR-NEW-EMULATOR-INPUT-COMPAT-20260806-001`
- 来源：`HANDOFF-YANBAO-LAOYANCAI-HIBOR-NEW-EMULATOR-CORE-INPUT-FIX-20260806-001`
- 开发负责人：`dev.developer.ana.cai`
- 审核入口：`dev.reviewer.ana.cai`
- 类型：需求支持的轻量缺陷修复；不修改公共采集合同。

## 实现

1. `AdbClient.input_text` 对 `[A-Za-z0-9._-]+` 使用 `adb shell input text`；现有 UI 层继续要求输入框文本与查询逐字相等。
2. 其他输入仅在 `ime list -s` 明确包含 `com.microvirt.memuime/.MemuIME` 时使用设备内 Base64 剪贴板服务；服务非零立即 fail closed，绝不发送 paste key。
3. `AdbClient.start_package` 仅在 `monkey` 明确缺失时回退到导出的 `cn.com.hibor/.WelcomeActivity`；其他 monkey 错误不被掩盖。
4. 缓存、PDF、hash、quota、manifest、选择与归档合同均未修改。

## 验收

- 定向回归：安全 ASCII、Memu 服务失败不 paste、monkey 缺失回退、非 ASCII/特殊输入在不支持环境停止、UI 逐字回读，共 `5/5 PASS`。
- 完整离线回归：`59/59 PASS`；`compileall=PASS`；`git diff --check=PASS`。
- 真实 APP、ADB、下载、额度、归档、网络、数据库动作均为 `0`。

## 当前状态

`IMPLEMENTED_PENDING_LIGHT_INDEPENDENT_REVIEW`

 dev/ana-dev/hibor_fast_collection/adb.py

@@ -40,6 +40,9 @@


BOUNDS_RE = re.compile(r"^\[(\d+),(\d+)\]\[(\d+),(\d+)\]$")
SAFE_ANDROID_INPUT_TEXT_RE = re.compile(r"^[A-Za-z0-9._-]+$")
MEMU_IME_COMPONENT = "com.microvirt.memuime/.MemuIME"
WELCOME_ACTIVITY = "cn.com.hibor/.WelcomeActivity"


class AdbClient:
@@ -197,25 +200,61 @@
    def input_text(self, value: str) -> None:
        if any(ch in value for ch in "\n\r\x00"):
            raise ContractError(ErrorCode.TASK_SPEC_INVALID, "query", "control character")
        # The real MEmu device uses MemuIME.  Android's generic ``input text``
        # drops non-ASCII search terms, so replace the focused field through the
        # emulator's UTF-8/Base64 clipboard service and paste key.  The repeated
        # keycodes are a single argv invocation (not a remote shell script).
        use_android_input = SAFE_ANDROID_INPUT_TEXT_RE.fullmatch(value) is not None
        if not use_android_input:
            ime_list = self.run("shell", "ime", "list", "-s")
            installed_imes = {
                line.strip()
                for line in ime_list.stdout_bytes.decode("utf-8", "replace").splitlines()
                if line.strip()
            }
            if ime_list.exit_code != 0 or MEMU_IME_COMPONENT not in installed_imes:
                raise ContractError(
                    ErrorCode.UI_ANCHOR_DRIFT, "input_text",
                    "safe input unavailable: MemuIME unsupported",
                )

        end = self.run("shell", "input", "keyevent", "123")
        clear = self.run("shell", "input", "keyevent", *("67" for _ in range(256)))
        if end.exit_code != 0 or clear.exit_code != 0:
            raise ContractError(ErrorCode.UI_ANCHOR_DRIFT, "input_text", "failed")

        if use_android_input:
            typed = self.run("shell", "input", "text", value)
            if typed.exit_code != 0:
                raise ContractError(ErrorCode.UI_ANCHOR_DRIFT, "input_text", "generic input failed")
            return

        encoded = base64.b64encode(value.encode("utf-8")).decode("ascii")
        clipboard = self.run(
            "shell", "am", "startservice", "-a", "memu_clip_board_base64",
            "--es", "clip", encoded,
        )
        if clipboard.exit_code != 0:
            # Never issue paste after a failed clipboard service call: on some
            # emulators that would paste unrelated host clipboard contents.
            raise ContractError(ErrorCode.UI_ANCHOR_DRIFT, "input_text", "Memu clipboard failed")
        paste = self.run("shell", "input", "keyevent", "279")
        if any(result.exit_code != 0 for result in (end, clear, clipboard, paste)):
            raise ContractError(ErrorCode.UI_ANCHOR_DRIFT, "input_text", "failed")
        if paste.exit_code != 0:
            raise ContractError(ErrorCode.UI_ANCHOR_DRIFT, "input_text", "paste failed")

    def start_package(self) -> None:
        result = self.run("shell", "monkey", "-p", self.package_name, "-c", "android.intent.category.LAUNCHER", "1")
        if result.exit_code != 0:
            raise ContractError(ErrorCode.PACKAGE_MISSING, "start_package", "failed")
        if result.exit_code == 0:
            return
        diagnostic = (result.stdout_bytes + b"\n" + result.stderr_bytes).decode("utf-8", "replace").casefold()
        monkey_missing = result.exit_code == 127 or (
            "monkey" in diagnostic and any(token in diagnostic for token in ("not found", "inaccessible"))
        )
        if not monkey_missing:
            raise ContractError(ErrorCode.PACKAGE_MISSING, "start_package", "monkey failed")
        fallback = self.run("shell", "am", "start", "-W", "-n", WELCOME_ACTIVITY)
        fallback_text = (fallback.stdout_bytes + b"\n" + fallback.stderr_bytes).decode(
            "utf-8", "replace"
        ).casefold()
        if (fallback.exit_code != 0 or "error:" in fallback_text
                or "error type" in fallback_text or "exception" in fallback_text):
            raise ContractError(ErrorCode.PACKAGE_MISSING, "start_package", "WelcomeActivity failed")

    def _require_cache_path(self, path: str) -> None:
        root = PurePosixPath(self.cache_root)

 dev/ana-dev/test/hibor_fast_collection/test_adb_input_compat.py

New file
@@ -0,0 +1,107 @@
from __future__ import annotations

from pathlib import Path
import unittest

from hibor_fast_collection.adb import AdbClient, UiNode
from hibor_fast_collection.models import ContractError, ErrorCode
from hibor_fast_collection.process import ProcessResult
from hibor_fast_collection.ui import HiborUiDriver, UiConfig


def result(exit_code: int = 0, *, stdout: bytes = b"", stderr: bytes = b"") -> ProcessResult:
    return ProcessResult(
        (), True, 1, exit_code, False, False, True, False,
        stdout, stderr, False, False, None, None, 1, None,
    )


class ScriptedSupervisor:
    def __init__(self, responses: list[ProcessResult]):
        self.responses = list(responses)
        self.calls: list[tuple[str, ...]] = []

    def run(self, argv, *, timeout_ms, **kwargs):
        self.calls.append(tuple(argv))
        if not self.responses:
            raise AssertionError(f"unexpected command: {argv!r}")
        return self.responses.pop(0)


def client(responses: list[ProcessResult]) -> tuple[AdbClient, ScriptedSupervisor]:
    supervisor = ScriptedSupervisor(responses)
    return AdbClient(Path("adb.exe"), supervisor, serial="emulator-test"), supervisor


class AdbInputCompatibilityTests(unittest.TestCase):
    def test_safe_ascii_uses_generic_input_without_clipboard_or_paste(self):
        adb, supervisor = client([result(), result(), result()])

        adb.input_text("603179")

        commands = [call[3:] for call in supervisor.calls]
        self.assertEqual(commands[-1], ("shell", "input", "text", "603179"))
        self.assertFalse(any("memu_clip_board_base64" in call for call in supervisor.calls))
        self.assertFalse(any(call[-2:] == ("keyevent", "279") for call in supervisor.calls))

    def test_memu_service_failure_never_pastes(self):
        adb, supervisor = client([
            result(stdout=b"com.microvirt.memuime/.MemuIME\n"),
            result(), result(), result(exit_code=1, stderr=b"service unavailable"),
        ])

        with self.assertRaisesRegex(ContractError, "Memu clipboard failed"):
            adb.input_text("\u56fd\u74f7\u6750\u6599")

        self.assertTrue(any("memu_clip_board_base64" in call for call in supervisor.calls))
        self.assertFalse(any(call[-2:] == ("keyevent", "279") for call in supervisor.calls))

    def test_monkey_missing_falls_back_to_exported_welcome_activity(self):
        adb, supervisor = client([
            result(exit_code=127, stderr=b"/system/bin/sh: monkey: not found"),
            result(stdout=b"Starting: Intent { cmp=cn.com.hibor/.WelcomeActivity }\nStatus: ok\n"),
        ])

        adb.start_package()

        self.assertEqual(
            supervisor.calls[-1][3:],
            ("shell", "am", "start", "-W", "-n", "cn.com.hibor/.WelcomeActivity"),
        )

    def test_unsupported_non_ascii_or_sensitive_input_fails_before_edit_or_paste(self):
        for value in ("\u4e09\u73af\u96c6\u56e2", "603179;paste"):
            with self.subTest(value=value):
                adb, supervisor = client([result(stdout=b"com.android.inputmethod.pinyin/.InputService\n")])
                with self.assertRaisesRegex(ContractError, "MemuIME unsupported") as raised:
                    adb.input_text(value)
                self.assertEqual(raised.exception.code, ErrorCode.UI_ANCHOR_DRIFT)
                self.assertEqual(len(supervisor.calls), 1)
                self.assertEqual(supervisor.calls[0][3:], ("shell", "ime", "list", "-s"))

    def test_ui_requires_exact_character_for_character_readback(self):
        field = UiNode("", "search-input", "android.widget.EditText", (0, 0, 10, 10), True)
        wrong = UiNode("603178", "search-input", "android.widget.EditText", (0, 0, 10, 10), True)

        class UiAdb:
            def __init__(self):
                self.snapshots = [(field,), (wrong,)]

            def ui_nodes(self):
                return self.snapshots.pop(0)

            def tap(self, *args):
                pass

            def input_text(self, value):
                self.value = value

        config = UiConfig((), "search-input", ("\u641c\u7d22",), "", (), "", "", "", "", (0, 0, 0, 0, 1))
        driver = HiborUiDriver(UiAdb(), config)

        with self.assertRaisesRegex(ContractError, "text mismatch"):
            driver.search("603179")


if __name__ == "__main__":
    unittest.main()