From aa378522965792775ccb79784700550238f5a6b2 Mon Sep 17 00:00:00 2001 From: Krzysztof kuhy Rudnicki Date: Fri, 25 Sep 2026 16:09:45 +0200 Subject: [PATCH] phone_guard: lease or block raw adb; android_ui drives a session's own virtual display A raw `adb shell input tap` took no lease, so one session's tap landed in another session's app (2026-09-25). python_pkg.phone_guard classifies every adb call in a Bash command and, as the Claude Code PreToolUse hook, takes the matching phone_lease (one app / real screen / whole phone) or exits 2: key/text input without -d, input on another session's virtual display, or a lease still held elsewhere after 40 s. android_ui gains --display self|: it drives only this session's phone_vd display under that app's lease, and every input carries -d. uiautomator sees display 0 only, so virtual-display trees come from a tiny instrumentation APK (A11yDump.java) built from source without Gradle. Prunes the oldest evidence artifact to stay under the 60-file cap. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01WpA6wW2DKiovQADTdjWJoH --- .../contracts/phone-guard-2026-09-25.json | 14 ++ .../evidence/phone-guard-2026-09-25.json | 18 ++ .../ubuntu-perf-fixes-tests-2026-08-23.json | 57 ----- python_pkg/android_ui/A11yDump.java | 133 +++++++++++ python_pkg/android_ui/README.md | 31 ++- python_pkg/android_ui/_a11y_helper.py | 225 ++++++++++++++++++ python_pkg/android_ui/_text_entry.py | 14 +- python_pkg/android_ui/a11y_dump_manifest.xml | 12 + python_pkg/android_ui/cli.py | 81 ++++++- python_pkg/android_ui/driver.py | 30 ++- python_pkg/android_ui/tests/conftest.py | 85 +++++++ .../android_ui/tests/test_a11y_helper.py | 205 ++++++++++++++++ python_pkg/android_ui/tests/test_cli.py | 24 +- .../android_ui/tests/test_cli_display.py | 83 +++++++ python_pkg/android_ui/tests/test_driver.py | 87 +++---- .../android_ui/tests/test_shell_helpers.py | 9 +- .../android_ui/tests/test_text_entry.py | 19 +- python_pkg/phone_guard/__init__.py | 9 + python_pkg/phone_guard/__main__.py | 10 + python_pkg/phone_guard/_need.py | 18 ++ python_pkg/phone_guard/_remote.py | 119 +++++++++ python_pkg/phone_guard/classify.py | 165 +++++++++++++ python_pkg/phone_guard/hook.py | 138 +++++++++++ python_pkg/phone_guard/tests/__init__.py | 1 + python_pkg/phone_guard/tests/conftest.py | 33 +++ python_pkg/phone_guard/tests/test_classify.py | 121 ++++++++++ python_pkg/phone_guard/tests/test_hook.py | 128 ++++++++++ .../phone_guard/tests/test_main_entry.py | 17 ++ python_pkg/phone_guard/tests/test_vd_state.py | 45 ++++ python_pkg/phone_guard/vd_state.py | 83 +++++++ 30 files changed, 1858 insertions(+), 156 deletions(-) create mode 100644 docs/superpowers/contracts/phone-guard-2026-09-25.json create mode 100644 docs/superpowers/evidence/phone-guard-2026-09-25.json delete mode 100644 docs/superpowers/evidence/ubuntu-perf-fixes-tests-2026-08-23.json create mode 100644 python_pkg/android_ui/A11yDump.java create mode 100644 python_pkg/android_ui/_a11y_helper.py create mode 100644 python_pkg/android_ui/a11y_dump_manifest.xml create mode 100644 python_pkg/android_ui/tests/conftest.py create mode 100644 python_pkg/android_ui/tests/test_a11y_helper.py create mode 100644 python_pkg/android_ui/tests/test_cli_display.py create mode 100644 python_pkg/phone_guard/__init__.py create mode 100644 python_pkg/phone_guard/__main__.py create mode 100644 python_pkg/phone_guard/_need.py create mode 100644 python_pkg/phone_guard/_remote.py create mode 100644 python_pkg/phone_guard/classify.py create mode 100644 python_pkg/phone_guard/hook.py create mode 100644 python_pkg/phone_guard/tests/__init__.py create mode 100644 python_pkg/phone_guard/tests/conftest.py create mode 100644 python_pkg/phone_guard/tests/test_classify.py create mode 100644 python_pkg/phone_guard/tests/test_hook.py create mode 100644 python_pkg/phone_guard/tests/test_main_entry.py create mode 100644 python_pkg/phone_guard/tests/test_vd_state.py create mode 100644 python_pkg/phone_guard/vd_state.py diff --git a/docs/superpowers/contracts/phone-guard-2026-09-25.json b/docs/superpowers/contracts/phone-guard-2026-09-25.json new file mode 100644 index 00000000..863e2e89 --- /dev/null +++ b/docs/superpowers/contracts/phone-guard-2026-09-25.json @@ -0,0 +1,14 @@ +{ + "title": "phone_guard: lease or block raw adb; android_ui drives a session's own virtual display", + "objective": "No Claude session's adb call can land in another session's app: raw adb in a Bash call takes the matching phone_lease or is blocked, and android_ui can tap by element on the session's own virtual display.", + "acceptance_criteria": [ + "Reads (logcat, dumpsys, pull, screencap) pass with no lease.", + "Input or launches on display 0 take the pkg:__screen__ lease; install/force-stop take that app's lease; reboot, Wi-Fi/airplane/Doze and settings put take the whole-phone lease.", + "input text/keyevent without -d, and input aimed at another session's virtual display, exit 2 with a reason.", + "A lease held by another session is waited on for 40 s, then the call exits 2 naming the holder.", + "android_ui --display self| drives only this session's display under that app's lease; every input call carries -d.", + "Virtual-display trees come from the com.kuhy.a11ydump instrumentation helper, built from source and installed with adb install -r -t." + ], + "out_of_scope": ["phone_lease scopes themselves (c0c51036)", "the ~/.claude hook wrapper, phone_deploy.sh and the phone-focus-mode whitelist (separate repos)"], + "verifier": "pytest python_pkg/android_ui python_pkg/phone_guard at 100% branch coverage; pre-commit on the changed files; a real helper APK build; the live two-session run on the Pixel follows this commit." +} diff --git a/docs/superpowers/evidence/phone-guard-2026-09-25.json b/docs/superpowers/evidence/phone-guard-2026-09-25.json new file mode 100644 index 00000000..d967ada0 --- /dev/null +++ b/docs/superpowers/evidence/phone-guard-2026-09-25.json @@ -0,0 +1,18 @@ +{ + "intent": "Several Claude sessions act on one phone at once without interrupting each other; a raw `adb shell input tap` (no lease) had landed in another session's app on 2026-09-25.", + "scope": ["python_pkg/phone_guard (new: classify, _remote, _need, vd_state, hook, CLI, tests)", "python_pkg/android_ui (--display, _a11y_helper + A11yDump.java, -d on every input, tests split under the 250-line cap)"], + "changes": [ + "phone_guard.classify maps every adb call in a shell command to a Need: nothing, screen, one app, whole phone, or a block", + "phone_guard.hook applies it as the Claude Code PreToolUse hook: acquires as this session, waits 40 s, exits 2 on refusal", + "android_ui --display self| resolves the session's phone_vd display and takes its package lease; a foreign display is refused (exit 5)", + "_a11y_helper builds a tiny instrumentation APK (javac, d8, aapt2, apksigner; no Gradle), cached by source hash, reading getWindowsOnAllDisplays()", + "The helper source sits flat beside the module (A11yDump.java, a11y_dump_manifest.xml) to respect the two-directory path cap" + ], + "verification": [ + {"command": "pytest python_pkg/android_ui python_pkg/phone_guard --cov --cov-branch", "result": "pass", "evidence": "152 passed; every android_ui and phone_guard module 100% line and branch"}, + {"command": "python3 -c 'from python_pkg.android_ui import _a11y_helper as h; print(h.build())'", "result": "pass", "evidence": "~/.cache/android_ui/a11ydump-fb2baa8888080a3c.apk built with the real SDK"}, + {"command": "pre-commit run --files ", "result": "pass", "evidence": "all hooks passed"} + ], + "risks": ["The hook passes unknown commands: it guards against interrupting other sessions, it is not a sandbox.", "The helper must stay in phone-focus-mode's whitelist or the focus-owner sweep hides it."], + "rollback": ["git revert the commit", "Remove the phone_guard_pretool.sh entry from ~/.claude/settings.json PreToolUse"] +} diff --git a/docs/superpowers/evidence/ubuntu-perf-fixes-tests-2026-08-23.json b/docs/superpowers/evidence/ubuntu-perf-fixes-tests-2026-08-23.json deleted file mode 100644 index 2d814a46..00000000 --- a/docs/superpowers/evidence/ubuntu-perf-fixes-tests-2026-08-23.json +++ /dev/null @@ -1,57 +0,0 @@ -{ - "intent": "Test ubuntu_perf_fixes.sh to 100% line coverage and take it off the shell coverage allowlist, leaving one untested lib in fixes/lib.", - "scope": [ - "linux_configuration/scripts/single_use/fixes/lib/ubuntu_perf_fixes.sh (four test seams)", - "linux_configuration/scripts/single_use/fixes/lib/tests/test_ubuntu_perf_fixes.sh and test_ubuntu_perf_fixes_oom.sh (new)", - "linux_configuration/scripts/single_use/fixes/lib/tests/ubuntu_perf_harness.sh (extended, not created -- it already existed)", - "meta/shell-coverage-allowlist.txt (one line removed)" - ], - "changes": [ - "Added SYSCTL_DROPIN_DIR, SYSTEMD_UNIT_DIR (twice) and EARLYOOM_CONF_FILE seams, each defaulting to the real path so production behaviour is unchanged", - "Deliberately left the add_undo lines with their literal /etc paths: the undo script runs on the real system, not in the sandbox, so seaming them would have written a broken undo", - "Extended the existing ubuntu_perf_harness.sh with the three new overrides, three more default stubs (nvidia-smi, nvidia-persistenced, dpkg) and the _t_unit / _t_sysctl_file readers", - "Added 59 assertions over 14 cases across two files covering all four functions", - "Removed ubuntu_perf_fixes.sh from meta/shell-coverage-allowlist.txt, taking it from 86 to 85 entries" - ], - "verification": [ - { - "command": "shell_coverage_jail.sh --timeout 600s --measure ubuntu_perf_fixes.sh --min 100 --fail-on-case-error", - "result": "pass", - "evidence": "ubuntu_perf_fixes.sh: 81/81 lines = 100.00%, exit 0, no case timeout." - }, - { - "command": "shell_coverage_jail.sh --timeout 600s --measure bt_audio.sh --min 100 (canary, with the two new test files present)", - "result": "pass", - "evidence": "bt_audio.sh: 127/127 lines = 100.00%. This is the lib a previous test file silently broke, so it is re-measured after every addition to this suite." - }, - { - "command": "bash meta/scripts/check_shell_coverage.sh .../ubuntu_perf_fixes.sh, AFTER deleting its allowlist line", - "result": "pass", - "evidence": "'every checked shell library meets the coverage bar', EXIT 0." - }, - { - "command": "ls -la /etc/sysctl.d/99-performance-tuning.conf /etc/default/earlyoom", - "result": "pass", - "evidence": "Both 'No such file or directory' after the suite ran, confirming the seams kept every write inside the throwaway tmpdir." - }, - { - "command": "bash linux_configuration/scripts/single_use/fixes/lib/tests/run_all.sh", - "result": "pass", - "evidence": "All fixes lib test files passed; 31 + 28 = 59 new assertions, 0 failed." - }, - { - "command": "shellcheck + shfmt on the lib, the harness and both test files", - "result": "pass", - "evidence": "No findings, zero suppressions. Lib is 219 lines, under the 250-line cap." - } - ], - "risks": [ - "fix_earlyoom writes a config on a machine that may already have one; the test asserts the .bak backup is taken before the overwrite, which is the behaviour that makes that safe.", - "The undo lines are asserted to contain the REAL /etc paths, so a future seam applied to them would fail these tests rather than silently producing an undo script that removes nothing.", - "fix_failed_sssd's detection loop breaks at the first failed unit while its masking loop re-checks each one; case 7 covers the single-failure arm specifically, since a naive implementation would mask all five." - ], - "rollback": [ - "git revert the commit, then re-add the ubuntu_perf_fixes.sh line to meta/shell-coverage-allowlist.txt", - "Re-run check_shell_coverage.sh to confirm the gate passes again with the exemption restored" - ] -} diff --git a/python_pkg/android_ui/A11yDump.java b/python_pkg/android_ui/A11yDump.java new file mode 100644 index 00000000..3e873181 --- /dev/null +++ b/python_pkg/android_ui/A11yDump.java @@ -0,0 +1,133 @@ +package com.kuhy.a11ydump; + +import android.accessibilityservice.AccessibilityServiceInfo; +import android.app.Activity; +import android.app.Instrumentation; +import android.app.UiAutomation; +import android.graphics.Rect; +import android.os.Bundle; +import android.os.SystemClock; +import android.util.SparseArray; +import android.view.accessibility.AccessibilityNodeInfo; +import android.view.accessibility.AccessibilityWindowInfo; +import java.util.List; + +/** + * Dumps one display's accessibility tree as uiautomator-style XML. + * + *

Run: {@code am instrument -w -e display com.kuhy.a11ydump/.A11yDump}. The + * XML comes back on one line as the {@code xml} result key, in the same shape + * {@code uiautomator dump} writes, so the Python side parses both with one + * parser. Coordinates are the display's own, which is what {@code input -d} + * expects. + */ +public final class A11yDump extends Instrumentation { + private static final int ATTEMPTS = 15; + private static final long RETRY_MS = 200; + + private Bundle arguments; + + @Override + public void onCreate(Bundle args) { + arguments = args; + start(); + } + + @Override + public void onStart() { + Bundle out = new Bundle(); + try { + int display = Integer.parseInt(arguments.getString("display", "0")); + UiAutomation automation = getUiAutomation( + UiAutomation.FLAG_DONT_SUPPRESS_ACCESSIBILITY_SERVICES); + AccessibilityServiceInfo info = automation.getServiceInfo(); + info.flags |= AccessibilityServiceInfo.FLAG_RETRIEVE_INTERACTIVE_WINDOWS + | AccessibilityServiceInfo.FLAG_INCLUDE_NOT_IMPORTANT_VIEWS + | AccessibilityServiceInfo.FLAG_REPORT_VIEW_IDS; + automation.setServiceInfo(info); + out.putString("xml", dump(automation, display)); + finish(Activity.RESULT_OK, out); + } catch (Throwable t) { + out.putString("error", String.valueOf(t)); + finish(Activity.RESULT_CANCELED, out); + } + } + + private static String dump(UiAutomation automation, int display) { + // Windows appear a beat after the service info changes; an empty + // list on the first read means "not yet", not "nothing there". + List windows = null; + for (int i = 0; i < ATTEMPTS; i++) { + SparseArray> all = + automation.getWindowsOnAllDisplays(); + windows = all.get(display); + if (windows != null && !windows.isEmpty()) break; + SystemClock.sleep(RETRY_MS); + } + StringBuilder xml = new StringBuilder( + "" + + ""); + if (windows != null) { + // Bottom layer first, like uiautomator, so a dialog's nodes come + // after (and so win over) the screen it covers. + for (int w = windows.size() - 1; w >= 0; w--) { + AccessibilityNodeInfo root = windows.get(w).getRoot(); + if (root != null) node(root, 0, xml); + } + } + return xml.append("").toString(); + } + + private static void node(AccessibilityNodeInfo n, int index, StringBuilder xml) { + Rect r = new Rect(); + n.getBoundsInScreen(r); + xml.append(""); + for (int i = 0; i < n.getChildCount(); i++) { + AccessibilityNodeInfo child = n.getChild(i); + if (child != null) node(child, i, xml); + } + xml.append(""); + } + + private static void flag(StringBuilder xml, String name, boolean value) { + xml.append(' ').append(name).append("=\"").append(value).append('"'); + } + + private static void attr(StringBuilder xml, String name, CharSequence value) { + xml.append(' ').append(name).append("=\""); + if (value != null) { + for (int i = 0; i < value.length(); i++) { + char c = value.charAt(i); + switch (c) { + case '&': xml.append("&"); break; + case '<': xml.append("<"); break; + case '>': xml.append(">"); break; + case '"': xml.append("""); break; + // Newlines would split the one-line result `am` prints. + case '\n': xml.append(" "); break; + case '\r': xml.append(" "); break; + default: xml.append(c); + } + } + } + xml.append('"'); + } +} diff --git a/python_pkg/android_ui/README.md b/python_pkg/android_ui/README.md index 1c5176e9..f2bfb285 100644 --- a/python_pkg/android_ui/README.md +++ b/python_pkg/android_ui/README.md @@ -90,9 +90,30 @@ ui.wait_for("Connected.", timeout=30) ## One phone, many sessions -Every command first takes this session's lease on the phone -(`python_pkg.phone_lease`): the owner is the nearest ancestor `claude` -process, the lease lasts 180 s past the last call, and a call from another +Sessions share the phone by working on separate displays. By default every +command drives the real screen (display 0) and takes this session's +`pkg:__screen__` lease (`python_pkg.phone_lease`: owner = the nearest +ancestor `claude` process, 180 s past the last call). A call from another session waits up to 60 s, then exits 3 naming the holder. Do not retry in a -loop — use the phone's virtual display (`~/.claude/scripts/phone_vd.sh`) or -the headless emulator (`~/.claude/scripts/phone_emu.sh`), or ask the user. +loop. + +`--display self` drives this session's virtual display instead +(`~/.claude/scripts/phone_vd.sh start ` creates it) and takes only +that app's lease, so several sessions can act at once: + + android-ui --display self dump + android-ui --display self tap "Save" + +`uiautomator dump` sees display 0 only, so on a virtual display the tree +comes from the `com.kuhy.a11ydump` instrumentation helper (`A11yDump.java`, +`_a11y_helper.py`): built once into `~/.cache/android_ui/` and installed +with `adb install -r -t` on first use. Its package must stay in +phone-focus-mode's whitelist, or the focus-owner sweep hides it. Every +`input` call carries `-d` (0 for the real screen), because input without it +goes to whichever display last took focus. A numeric `--display` that is +another session's virtual display is refused. + +Raw `adb` from a Claude Bash call is covered too: the +`~/.claude/hooks/phone_guard_pretool.sh` hook (`python_pkg.phone_guard`) +takes the matching lease or blocks the call. See the `phone-deploy` skill, +section 3, for the full rules. diff --git a/python_pkg/android_ui/_a11y_helper.py b/python_pkg/android_ui/_a11y_helper.py new file mode 100644 index 00000000..126ca038 --- /dev/null +++ b/python_pkg/android_ui/_a11y_helper.py @@ -0,0 +1,225 @@ +"""Accessibility-tree dumps of any display, via a tiny instrumentation APK. + +``uiautomator dump`` only ever sees display 0, so an app on a session's +virtual display (``phone_vd.sh``) could only be tapped by coordinates. The +APK built from ``A11yDump.java`` runs under ``am instrument``, where a UiAutomation can +read ``getWindowsOnAllDisplays()`` and print one display's tree in the same +XML shape ``uiautomator`` writes. + +It is built from source here (javac → d8 → aapt2 → apksigner, no Gradle), +cached by source hash under ``~/.cache/android_ui/``, and (re)installed with +``adb install -r -t`` whenever the phone holds a different build. The APK is +never committed: the repo's binary gate forbids it, and the source is enough. +""" + +from __future__ import annotations + +from collections.abc import Callable +import hashlib +import os +from pathlib import Path +import shutil +import subprocess +import tempfile +import zipfile + +from python_pkg.android_ui._elements import UiAutomationError + +PACKAGE = "com.kuhy.a11ydump" +_RUNNER = f"{PACKAGE}/.A11yDump" +# Flat beside this module: the repo caps paths at two directories, and javac +# does not need the package's directory layout for an explicitly named file. +_JAVA = Path(__file__).with_name("A11yDump.java") +_MANIFEST = Path(__file__).with_name("a11y_dump_manifest.xml") +_MIN_API = "30" # getWindowsOnAllDisplays() + +# (adb args...) -> stdout, raising UiAutomationError on failure. +AdbRun = Callable[..., str] + + +def source_hash() -> str: + """Hash of every source file: the build's identity and versionName.""" + digest = hashlib.sha256() + for path in (_JAVA, _MANIFEST): + digest.update(path.name.encode()) + digest.update(path.read_bytes()) + return digest.hexdigest()[:16] + + +def cache_dir() -> Path: + """Where built APKs are kept.""" + return Path( + os.environ.get("ANDROID_UI_CACHE", Path.home() / ".cache" / "android_ui") + ) + + +def sdk_root() -> Path: + """The Android SDK, from $ANDROID_HOME / $ANDROID_SDK_ROOT.""" + for var in ("ANDROID_HOME", "ANDROID_SDK_ROOT"): + value = os.environ.get(var) + if value and Path(value).is_dir(): + return Path(value) + fallback = Path.home() / "sdk" / "Android" / "Sdk" + if fallback.is_dir(): + return fallback + msg = "no Android SDK: set ANDROID_HOME (needs build-tools and a platform)" + raise UiAutomationError(msg) + + +def _newest(parent: Path, prefix: str = "") -> Path: + candidates = sorted( + (p for p in parent.iterdir() if p.name.startswith(prefix) and p.is_dir()), + key=lambda p: [ + int(x) if x.isdigit() else 0 for x in p.name.split("-")[-1].split(".") + ], + ) + if not candidates: + msg = f"nothing matching {prefix!r} under {parent}" + raise UiAutomationError(msg) + return candidates[-1] + + +def build_tool(name: str) -> Path: + """Absolute path of an SDK build tool (``aapt2``, ``d8``, ...), newest version.""" + return _newest(sdk_root() / "build-tools") / name + + +def _tool(*cmd: str | Path) -> None: + done = subprocess.run( + [str(c) for c in cmd], capture_output=True, text=True, check=False + ) + if done.returncode != 0: + msg = f"building the a11y helper failed: {cmd[0]}\n{done.stderr.strip()}" + raise UiAutomationError(msg) + + +def build() -> Path: + """Return the helper APK for the current source, building it if needed.""" + version = source_hash() + apk = cache_dir() / f"a11ydump-{version}.apk" + if apk.is_file(): + return apk + sdk = sdk_root() + tools = _newest(sdk / "build-tools") + android_jar = _newest(sdk / "platforms", "android-") / "android.jar" + keystore = Path.home() / ".android" / "debug.keystore" + with tempfile.TemporaryDirectory() as tmp: + work = Path(tmp) + classes, dex = work / "classes", work / "dex" + classes.mkdir() + dex.mkdir() + _tool( + "javac", + "--release", + "11", + "-classpath", + android_jar, + "-d", + classes, + _JAVA, + ) + class_files = sorted(str(p) for p in classes.rglob("*.class")) + _tool( + tools / "d8", + "--release", + "--min-api", + _MIN_API, + "--lib", + android_jar, + "--output", + dex, + *class_files, + ) + unsigned = work / "unsigned.apk" + _tool( + tools / "aapt2", + "link", + "--manifest", + _MANIFEST, + "-I", + android_jar, + "--version-code", + "1", + "--version-name", + version, + "-o", + unsigned, + ) + with zipfile.ZipFile(unsigned, "a") as zf: + zf.write(dex / "classes.dex", "classes.dex") + aligned = work / "aligned.apk" + _tool(tools / "zipalign", "-f", "4", unsigned, aligned) + if not keystore.is_file(): + _tool( + "keytool", + "-genkeypair", + "-keystore", + keystore, + "-storepass", + "android", + "-alias", + "androiddebugkey", + "-keypass", + "android", + "-keyalg", + "RSA", + "-validity", + "10000", + "-dname", + "CN=Android Debug,O=Android,C=US", + ) + _tool( + tools / "apksigner", + "sign", + "--ks", + keystore, + "--ks-pass", + "pass:android", + "--key-pass", + "pass:android", + "--out", + work / "signed.apk", + aligned, + ) + apk.parent.mkdir(parents=True, exist_ok=True) + shutil.move(work / "signed.apk", apk) + return apk + + +def ensure_installed(run: AdbRun) -> None: + """Install the helper unless the phone already holds this exact build.""" + version = source_hash() + info = run("shell", "dumpsys", "package", PACKAGE) + if f"versionName={version}" in info: + return + run("install", "-r", "-t", str(build()), timeout=120.0) + + +def parse_result(output: str) -> str: + """Pull the XML out of ``am instrument -w`` output, or raise its error.""" + for line in output.splitlines(): + if line.startswith("INSTRUMENTATION_RESULT: xml="): + return line.split("=", 1)[1] + for line in output.splitlines(): + if line.startswith("INSTRUMENTATION_RESULT: error="): + msg = f"a11y helper failed: {line.split('=', 1)[1]}" + raise UiAutomationError(msg) + msg = f"a11y helper returned no tree:\n{output.strip()[-500:]}" + raise UiAutomationError(msg) + + +def dump_display(run: AdbRun, display: int) -> str: + """Return ``display``'s accessibility tree as uiautomator-style XML.""" + ensure_installed(run) + output = run( + "shell", + "am", + "instrument", + "-w", + "-e", + "display", + str(display), + _RUNNER, + timeout=45.0, + ) + return parse_result(output) diff --git a/python_pkg/android_ui/_text_entry.py b/python_pkg/android_ui/_text_entry.py index fa49c233..dead1d02 100644 --- a/python_pkg/android_ui/_text_entry.py +++ b/python_pkg/android_ui/_text_entry.py @@ -33,6 +33,10 @@ class TextEntryMixin(ABC): def _run(self, *args: str, timeout: float = 30.0) -> str: """Supplied by the concrete driver: run one adb command.""" + @abstractmethod + def _input(self, *args: str) -> str: + """Supplied by the concrete driver: ``input`` on the driven display.""" + @abstractmethod def dump(self, *, retries: int = 4) -> list[UiElement]: """Supplied by the concrete driver: read the current UI tree.""" @@ -49,9 +53,9 @@ class TextEntryMixin(ABC): return # KEYCODE_MOVE_END (123) then a run of deletes is more reliable across # IMEs than CTRL+A, which not every keyboard honours. - self._run("shell", "input", "keyevent", "123") + self._input("keyevent", "123") for _ in range(length + 2): - self._run("shell", "input", "keyevent", "67") + self._input("keyevent", "67") time.sleep(self._settle) def editable_fields(self) -> list[UiElement]: @@ -78,14 +82,14 @@ class TextEntryMixin(ABC): target = fields[index] before = target.text x, y = target.center - self._run("shell", "input", "tap", str(x), str(y)) + self._input("tap", str(x), str(y)) time.sleep(max(self._settle, 0.8)) # REPLACE, don't append. `input text` inserts at the cursor, so typing # into a field that already holds something silently concatenates -- # producing e.g. "old@example.comnew@example.com", which is accepted by # the widget, passes a "did the text change?" check, and is wrong. self._clear_focused_field(len(before)) - self._run("shell", "input", "text", _escape(text)) + self._input("text", _escape(text)) time.sleep(self._settle) after = self.editable_fields() changed = any( @@ -118,7 +122,7 @@ class TextEntryMixin(ABC): for keyevent in ("111", "4"): if not self.keyboard_is_up(): return - self._run("shell", "input", "keyevent", keyevent) + self._input("keyevent", keyevent) time.sleep(max(self._settle, 0.6)) if self.keyboard_is_up(): msg = ( diff --git a/python_pkg/android_ui/a11y_dump_manifest.xml b/python_pkg/android_ui/a11y_dump_manifest.xml new file mode 100644 index 00000000..5041580d --- /dev/null +++ b/python_pkg/android_ui/a11y_dump_manifest.xml @@ -0,0 +1,12 @@ + + + + + + + diff --git a/python_pkg/android_ui/cli.py b/python_pkg/android_ui/cli.py index b0fe5a95..a13ad3ca 100644 --- a/python_pkg/android_ui/cli.py +++ b/python_pkg/android_ui/cli.py @@ -9,6 +9,7 @@ successful no-op instead of guessing from a screenshot. android-ui tap "Connect Firebase" android-ui type "Sync account email" kuhy@example.com android-ui wait "Connected to Firebase." --timeout 30 + android-ui --display self tap "Save" # this session's phone_vd display """ from __future__ import annotations @@ -17,7 +18,24 @@ import argparse import sys from python_pkg.android_ui.driver import AndroidUi, UiAutomationError -from python_pkg.phone_lease import PhoneBusyError, acquire +from python_pkg.phone_guard.vd_state import display_owner, own_display +from python_pkg.phone_lease import ( + NoPhoneError, + PhoneBusyError, + acquire, + owner_id, + resolve_serial, + scope_of, +) + +# Exit codes shared with phone_lease: 3 held elsewhere, 4 no phone. +_EXIT_BUSY = 3 +_EXIT_NO_PHONE = 4 +_EXIT_BAD_DISPLAY = 5 + + +class _TargetError(RuntimeError): + """The requested display cannot be driven by this session.""" def _build_parser() -> argparse.ArgumentParser: @@ -27,6 +45,11 @@ def _build_parser() -> argparse.ArgumentParser: description="Drive an Android app by element, never by coordinates.", ) parser.add_argument("-s", "--serial", help="target device serial") + parser.add_argument( + "--display", + help="drive a virtual display instead of the real screen: its logical " + "id, or 'self' for the one phone_vd.sh started for this session", + ) parser.add_argument( "--exact", action="store_true", @@ -56,16 +79,43 @@ def _build_parser() -> argparse.ArgumentParser: return parser -def _lease_phone(serial: str | None, command: str) -> bool: - """Refresh this session's lease on the phone; False when another holds it. +def _target(serial: str, requested: str | None) -> tuple[int | None, str]: + """Return (display, lease scope) for ``--display``. - Every command goes through here first, so a second session's taps wait - instead of landing in whatever this one has in front (2026-09-20: a - sandbox tap opened another session's manga reader). Serial-less calls - share one key: adb picks one device anyway. + The real screen is shared, so it takes the screen lease. A virtual display + must be this session's own (another session's app would be driven blind) + and takes the lease on its app instead, which is what lets two sessions + work at the same time. + """ + if requested is None or requested == "0": + return None, scope_of() # the real screen: pkg:__screen__ + mine = own_display(serial, owner_id()) + if requested == "self": + if mine is None: + msg = "no virtual display for this session: run phone_vd.sh start " + raise _TargetError(msg) + return mine.display, scope_of(mine.package) + if not requested.isdigit(): + msg = f"--display takes a display id or 'self', not {requested!r}" + raise _TargetError(msg) + display = int(requested) + if mine is None or mine.display != display: + other = display_owner(serial, display) + whose = f"owned by {other.owner_dir}" if other else "not this session's" + msg = f"display {display} is {whose}; drive only your own (--display self)" + raise _TargetError(msg) + return display, scope_of(mine.package) + + +def _lease_phone(serial: str, command: str, scope: str) -> bool: + """Refresh this session's lease; False when another session holds it. + + Keyed by the resolved serial, never a placeholder: serial-less calls used + to lease "default" while phone_deploy leased the real serial, so neither + saw the other and one session's tap landed in another's app (2026-09-25). """ try: - acquire(serial or "default", f"android_ui {command}") + acquire(serial, f"android_ui {command}", scope=scope) except PhoneBusyError as exc: sys.stderr.write(f"android-ui: {exc}\n") return False @@ -97,10 +147,19 @@ def _run(ui: AndroidUi, args: argparse.Namespace) -> None: def main(argv: list[str] | None = None) -> int: """Run the CLI. Returns a process exit code.""" args = _build_parser().parse_args(argv) - if not _lease_phone(args.serial, args.command): - return 3 try: - _run(AndroidUi(serial=args.serial), args) + serial = resolve_serial(args.serial) + display, scope = _target(serial, args.display) + except NoPhoneError as exc: + sys.stderr.write(f"android-ui: {exc}\n") + return _EXIT_NO_PHONE + except _TargetError as exc: + sys.stderr.write(f"android-ui: {exc}\n") + return _EXIT_BAD_DISPLAY + if not _lease_phone(serial, args.command, scope): + return _EXIT_BUSY + try: + _run(AndroidUi(serial=serial, display=display), args) except UiAutomationError as exc: # Loud and non-zero on purpose: a silent no-op here is exactly the # failure this package exists to eliminate. The message already names diff --git a/python_pkg/android_ui/driver.py b/python_pkg/android_ui/driver.py index 2f038d53..3216bab6 100644 --- a/python_pkg/android_ui/driver.py +++ b/python_pkg/android_ui/driver.py @@ -8,6 +8,7 @@ import subprocess import tempfile import time +from python_pkg.android_ui import _a11y_helper from python_pkg.android_ui._elements import ( AmbiguousElementError, ElementNotFoundError, @@ -38,11 +39,19 @@ class AndroidUi(TextEntryMixin): *, adb: str = "adb", settle_seconds: float = 0.4, + display: int | None = None, ) -> None: - """Create a driver. ``serial`` targets one of several devices.""" + """Create a driver. ``serial`` targets one of several devices. + + ``display`` drives one display (a session's virtual display from + ``phone_vd.sh``) instead of the real screen: input goes to it with + ``input -d`` and its tree comes from the a11y helper, because + ``uiautomator dump`` cannot see any display but 0. + """ self._adb = adb self._serial = serial self._settle = settle_seconds + self._display = display def _run(self, *args: str, timeout: float = 30.0) -> str: """Run an adb command, returning stdout.""" @@ -69,6 +78,17 @@ class AndroidUi(TextEntryMixin): raise UiAutomationError(msg) return done.stdout + def _input(self, *args: str) -> str: + """Run ``input`` on the display this driver targets. + + Always with ``-d``, 0 meaning the real screen: without it, key and + text events go to whichever display last had focus, and a session's + virtual display takes focus the moment its app launches -- so another + session's typing lands in that app. + """ + display = 0 if self._display is None else self._display + return self._run("shell", "input", "-d", str(display), *args) + # ── Reading the screen ──────────────────────────────────────────────── def dump(self, *, retries: int = 4) -> list[UiElement]: @@ -96,6 +116,8 @@ class AndroidUi(TextEntryMixin): def _dump_once(self) -> list[UiElement]: """Pull one accessibility-tree snapshot.""" + if self._display is not None: + return _parse_tree(_a11y_helper.dump_display(self._run, self._display)) self._run("shell", "uiautomator", "dump", _REMOTE_DUMP, timeout=45.0) with tempfile.TemporaryDirectory() as tmp: local = Path(tmp) / "ui.xml" @@ -161,7 +183,7 @@ class AndroidUi(TextEntryMixin): self.dismiss_keyboard() element = self.wait_for(query, timeout=timeout, exact=exact) x, y = element.center - self._run("shell", "input", "tap", str(x), str(y)) + self._input("tap", str(x), str(y)) time.sleep(self._settle) return element @@ -182,10 +204,10 @@ class AndroidUi(TextEntryMixin): element = self.wait_for(query, timeout=timeout, exact=exact) before = element.text x, y = element.center - self._run("shell", "input", "tap", str(x), str(y)) + self._input("tap", str(x), str(y)) # The field has to actually take focus before keystrokes mean anything. time.sleep(max(self._settle, 0.8)) - self._run("shell", "input", "text", _escape(text)) + self._input("text", _escape(text)) time.sleep(self._settle) for candidate in self.dump(): diff --git a/python_pkg/android_ui/tests/conftest.py b/python_pkg/android_ui/tests/conftest.py new file mode 100644 index 00000000..9703d433 --- /dev/null +++ b/python_pkg/android_ui/tests/conftest.py @@ -0,0 +1,85 @@ +"""Shared fixtures and a scripted adb stand-in for the android_ui tests.""" + +from __future__ import annotations + +from pathlib import Path +from typing import TYPE_CHECKING +from unittest.mock import patch + +import pytest + +if TYPE_CHECKING: + from collections.abc import Iterator + +MOD = "python_pkg.android_ui.cli" + + +@pytest.fixture +def phone() -> Iterator[None]: + """One resolved phone, and no virtual display unless a test adds one.""" + with ( + patch(f"{MOD}.resolve_serial", side_effect=lambda s=None: s or "SER1"), + patch(f"{MOD}.owner_id", return_value="claude:1"), + patch(f"{MOD}.own_display", return_value=None), + patch(f"{MOD}.display_owner", return_value=None), + ): + yield + + +def node(text: str = "", **attrs: str) -> str: + """Render one ```` element for a fake dump. + + ``attrs`` accepts ``desc``, ``hint``, ``res``, ``cls``, ``bounds``, + ``enabled`` and ``focused``; each falls back to a representative default. + """ + return ( + f'' + ) + + +def tree(*nodes: str) -> str: + """Wrap ``nodes`` in a hierarchy document.""" + return f"{''.join(nodes)}" + + +class FakeDevice: + """Scripted stand-in for adb, recording every command it is given.""" + + def __init__(self, trees: list[str] | None = None) -> None: + self.trees = trees or [tree(node("Connect"))] + self.calls: list[tuple[str, ...]] = [] + self.keyboard_shown = False + self.pulled = 0 + + def run(self, *args: str, timeout: float = 30.0) -> str: + """Answer an adb invocation.""" + del timeout + self.calls.append(args) + if args[:2] == ("shell", "dumpsys") and args[2] == "input_method": + return f"mInputShown={'true' if self.keyboard_shown else 'false'}" + if args[:2] == ("shell", "dumpsys") and args[2] == "window": + return "mCurrentFocus=Window{ab12 u0 com.example/com.example.Main}" + if args[0] == "pull": + index = min(self.pulled, len(self.trees) - 1) + self.pulled += 1 + Path(args[2]).write_text(self.trees[index], encoding="utf-8") + if args[:5] == ("shell", "input", "-d", "0", "keyevent") and args[5] in { + "111", + "4", + }: + self.keyboard_shown = False + return "" + + def taps(self) -> list[tuple[int, int]]: + """Return every tap coordinate, in order.""" + return [ + (int(c[5]), int(c[6])) + for c in self.calls + if c[:5] == ("shell", "input", "-d", "0", "tap") + ] diff --git a/python_pkg/android_ui/tests/test_a11y_helper.py b/python_pkg/android_ui/tests/test_a11y_helper.py new file mode 100644 index 00000000..7b9c1f96 --- /dev/null +++ b/python_pkg/android_ui/tests/test_a11y_helper.py @@ -0,0 +1,205 @@ +"""Tests for the a11y dump helper: build caching, install, result parsing. + +The SDK tools are faked: each fake writes the file the real tool would, so the +build pipeline's wiring (what feeds what, what gets cached) is what is tested. +The real build and a real dump were verified on the Pixel 6a on 2026-09-25. +""" + +from __future__ import annotations + +from pathlib import Path +import shutil +import subprocess +from typing import TYPE_CHECKING +from unittest.mock import MagicMock, patch +import zipfile + +import pytest + +from python_pkg.android_ui import _a11y_helper as helper +from python_pkg.android_ui._elements import UiAutomationError + +if TYPE_CHECKING: + from collections.abc import Iterator + +MOD = "python_pkg.android_ui._a11y_helper" + + +@pytest.fixture +def source(tmp_path: Path) -> Iterator[Path]: + """A tiny helper source: one Java file and its manifest.""" + java = tmp_path / "A11yDump.java" + java.write_text("class A11yDump {}") + manifest = tmp_path / "a11y_dump_manifest.xml" + manifest.write_text("") + with patch(f"{MOD}._JAVA", java), patch(f"{MOD}._MANIFEST", manifest): + yield tmp_path + + +@pytest.fixture +def sdk(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + """A fake SDK layout, with the cache and HOME inside tmp_path.""" + root = tmp_path / "sdk" + for d in ( + "build-tools/35.0.0", + "build-tools/36.0.0", + "platforms/android-9", + "platforms/android-36", + "platforms/android-37.2", + ): + (root / d).mkdir(parents=True) + monkeypatch.setenv("ANDROID_HOME", str(root)) + monkeypatch.setenv("ANDROID_UI_CACHE", str(tmp_path / "cache")) + monkeypatch.setenv("HOME", str(tmp_path / "home")) + return root + + +def fake_tool(calls: list[list[str]]) -> MagicMock: + """subprocess.run stand-in that produces each tool's output file.""" + + def run(cmd: list[str], **_: object) -> subprocess.CompletedProcess[str]: + calls.append(cmd) + name = Path(cmd[0]).name + if name == "d8": + out = Path(cmd[cmd.index("--output") + 1]) + (out / "classes.dex").write_bytes(b"dex") + elif name == "aapt2": + with zipfile.ZipFile(cmd[cmd.index("-o") + 1], "w") as zf: + zf.writestr("AndroidManifest.xml", "m") + elif name == "zipalign": + shutil.copy(cmd[-2], cmd[-1]) + elif name == "apksigner": + shutil.copy(cmd[-1], cmd[cmd.index("--out") + 1]) + elif name == "keytool": + Path(cmd[cmd.index("-keystore") + 1]).parent.mkdir(parents=True) + Path(cmd[cmd.index("-keystore") + 1]).write_bytes(b"ks") + return subprocess.CompletedProcess(cmd, 0, "", "") + + return MagicMock(side_effect=run) + + +class TestIdentity: + def test_hash_follows_the_source(self, source: Path) -> None: + first = helper.source_hash() + assert first == helper.source_hash() + (source / "a11y_dump_manifest.xml").write_text("") + assert helper.source_hash() != first + + def test_cache_dir_defaults_under_home( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.delenv("ANDROID_UI_CACHE", raising=False) + assert helper.cache_dir() == Path.home() / ".cache" / "android_ui" + + +class TestSdk: + def test_env_wins(self, sdk: Path) -> None: + assert helper.sdk_root() == sdk + + def test_falls_back_to_home_sdk( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.delenv("ANDROID_HOME", raising=False) + monkeypatch.setenv("ANDROID_SDK_ROOT", str(tmp_path / "missing")) + fallback = tmp_path / "sdk" / "Android" / "Sdk" + fallback.mkdir(parents=True) + with patch(f"{MOD}.Path.home", return_value=tmp_path): + assert helper.sdk_root() == fallback + + def test_no_sdk_raises( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.delenv("ANDROID_HOME", raising=False) + monkeypatch.delenv("ANDROID_SDK_ROOT", raising=False) + with ( + patch(f"{MOD}.Path.home", return_value=tmp_path), + pytest.raises(UiAutomationError, match="no Android SDK"), + ): + helper.sdk_root() + + def test_build_tool_is_in_the_newest_build_tools(self, sdk: Path) -> None: + assert helper.build_tool("aapt2") == sdk / "build-tools" / "36.0.0" / "aapt2" + + def test_newest_version_wins_and_empty_raises(self, sdk: Path) -> None: + assert helper._newest(sdk / "build-tools").name == "36.0.0" + assert helper._newest(sdk / "platforms", "android-").name == "android-37.2" + with pytest.raises(UiAutomationError, match="nothing matching"): + helper._newest(sdk / "platforms", "nope-") + + +class TestBuild: + def test_builds_signs_and_caches(self, source: Path, sdk: Path) -> None: + del source, sdk + calls: list[list[str]] = [] + with patch(f"{MOD}.subprocess.run", fake_tool(calls)): + apk = helper.build() + assert apk.name == f"a11ydump-{helper.source_hash()}.apk" + with zipfile.ZipFile(apk) as zf: + assert "classes.dex" in zf.namelist() + tools = [Path(c[0]).name for c in calls] + assert tools == ["javac", "d8", "aapt2", "zipalign", "keytool", "apksigner"] + assert f"{Path.home()}/.android/debug.keystore" in calls[4] + calls.clear() + assert helper.build() == apk + assert calls == [], "a cached build runs no tool" + + def test_an_existing_keystore_is_reused(self, source: Path, sdk: Path) -> None: + del source, sdk + keystore = Path.home() / ".android" / "debug.keystore" + keystore.parent.mkdir(parents=True) + keystore.write_bytes(b"ks") + calls: list[list[str]] = [] + with patch(f"{MOD}.subprocess.run", fake_tool(calls)): + helper.build() + assert "keytool" not in [Path(c[0]).name for c in calls] + + def test_a_failing_tool_is_reported(self, source: Path, sdk: Path) -> None: + del source, sdk + failed = subprocess.CompletedProcess(["javac"], 1, "", "boom") + with ( + patch(f"{MOD}.subprocess.run", return_value=failed), + pytest.raises(UiAutomationError, match=r"javac\nboom"), + ): + helper.build() + + +class TestDevice: + def test_skips_install_when_the_phone_has_this_build(self, source: Path) -> None: + del source + run = MagicMock(return_value=f"versionName={helper.source_hash()}\n") + helper.ensure_installed(run) + run.assert_called_once_with("shell", "dumpsys", "package", helper.PACKAGE) + + def test_installs_when_missing_or_stale(self, source: Path, tmp_path: Path) -> None: + del source + run = MagicMock(return_value="versionName=old\n") + with patch(f"{MOD}.build", return_value=tmp_path / "a.apk"): + helper.ensure_installed(run) + run.assert_called_with( + "install", "-r", "-t", str(tmp_path / "a.apk"), timeout=120.0 + ) + + def test_parse_result(self) -> None: + out = "INSTRUMENTATION_RESULT: xml=\nINSTRUMENTATION_CODE: -1\n" + assert helper.parse_result(out) == "" + with pytest.raises(UiAutomationError, match=r"failed: java\.lang\.X"): + helper.parse_result("INSTRUMENTATION_RESULT: error=java.lang.X\n") + with pytest.raises(UiAutomationError, match="no tree"): + helper.parse_result("INSTRUMENTATION_STATUS: nothing\n") + + def test_dump_display(self) -> None: + run = MagicMock(return_value="INSTRUMENTATION_RESULT: xml=\n") + with patch(f"{MOD}.ensure_installed") as ensure: + assert helper.dump_display(run, 9) == "" + ensure.assert_called_once_with(run) + run.assert_called_once_with( + "shell", + "am", + "instrument", + "-w", + "-e", + "display", + "9", + "com.kuhy.a11ydump/.A11yDump", + timeout=45.0, + ) diff --git a/python_pkg/android_ui/tests/test_cli.py b/python_pkg/android_ui/tests/test_cli.py index 2261d98b..4df04abf 100644 --- a/python_pkg/android_ui/tests/test_cli.py +++ b/python_pkg/android_ui/tests/test_cli.py @@ -17,12 +17,15 @@ from python_pkg.android_ui.driver import ( ElementNotFoundError, UiElement, ) -from python_pkg.phone_lease import PhoneBusyError +from python_pkg.phone_lease import NoPhoneError, PhoneBusyError + +SCREEN_SCOPE = "pkg:__screen__" if TYPE_CHECKING: from collections.abc import Iterator MOD = "python_pkg.android_ui.cli" +pytestmark = pytest.mark.usefixtures("phone") def _element(label: str = "Connect Firebase") -> UiElement: @@ -48,15 +51,22 @@ def ui() -> Iterator[MagicMock]: class TestLease: - def test_every_command_refreshes_the_lease_first(self) -> None: - with patch(f"{MOD}.AndroidUi"), patch(f"{MOD}.acquire") as acquire: - assert cli.main(["-s", "SER1", "focus"]) == 0 - acquire.assert_called_once_with("SER1", "android_ui focus") + def test_every_command_refreshes_the_screen_lease_first(self) -> None: + with patch(f"{MOD}.AndroidUi") as factory, patch(f"{MOD}.acquire") as acquire: + assert cli.main(["-s", "SER2", "focus"]) == 0 + acquire.assert_called_once_with("SER2", "android_ui focus", scope=SCREEN_SCOPE) + factory.assert_called_once_with(serial="SER2", display=None) - def test_serial_less_calls_share_one_key(self) -> None: + def test_serial_less_calls_lease_the_resolved_serial(self) -> None: + # Leasing a "default" key let two sessions drive one phone blind. with patch(f"{MOD}.AndroidUi"), patch(f"{MOD}.acquire") as acquire: cli.main(["focus"]) - assert acquire.call_args.args[0] == "default" + assert acquire.call_args.args[0] == "SER1" + + def test_no_phone_exits_4(self, capsys: pytest.CaptureFixture[str]) -> None: + with patch(f"{MOD}.resolve_serial", side_effect=NoPhoneError("none")): + assert cli.main(["focus"]) == 4 + assert "none" in capsys.readouterr().err def test_a_foreign_lease_stops_the_command( self, capsys: pytest.CaptureFixture[str] diff --git a/python_pkg/android_ui/tests/test_cli_display.py b/python_pkg/android_ui/tests/test_cli_display.py new file mode 100644 index 00000000..7070b548 --- /dev/null +++ b/python_pkg/android_ui/tests/test_cli_display.py @@ -0,0 +1,83 @@ +"""Tests for ``android_ui --display``: which display a call may drive. + +The real screen is shared, so it takes the screen lease; a session's own +virtual display takes only its app's lease; anyone else's display is refused +before the driver or a lease is touched. +""" + +from __future__ import annotations + +from unittest.mock import patch + +import pytest + +from python_pkg.android_ui import cli +from python_pkg.phone_guard.vd_state import SessionDisplay + +MOD = "python_pkg.android_ui.cli" +SCREEN_SCOPE = "pkg:__screen__" +pytestmark = pytest.mark.usefixtures("phone") + + +def _mine(display: int = 9, package: str = "dev.kuhy.todo") -> SessionDisplay: + return SessionDisplay("claude_1", display, "sf", package) + + +class TestDisplay: + def test_self_drives_this_sessions_display_under_its_app_lease(self) -> None: + with ( + patch(f"{MOD}.own_display", return_value=_mine()), + patch(f"{MOD}.AndroidUi") as factory, + patch(f"{MOD}.acquire") as acquire, + ): + assert cli.main(["--display", "self", "focus"]) == 0 + factory.assert_called_once_with(serial="SER1", display=9) + assert acquire.call_args.kwargs["scope"] == "pkg:dev.kuhy.todo" + + def test_an_explicit_id_of_ones_own_display_works(self) -> None: + with ( + patch(f"{MOD}.own_display", return_value=_mine()), + patch(f"{MOD}.AndroidUi") as factory, + patch(f"{MOD}.acquire"), + ): + assert cli.main(["--display", "9", "focus"]) == 0 + assert factory.call_args.kwargs["display"] == 9 + + def test_display_0_is_the_real_screen(self) -> None: + with patch(f"{MOD}.AndroidUi") as factory, patch(f"{MOD}.acquire") as acquire: + assert cli.main(["--display", "0", "focus"]) == 0 + assert factory.call_args.kwargs["display"] is None + assert acquire.call_args.kwargs["scope"] == SCREEN_SCOPE + + @pytest.mark.parametrize( + ("requested", "mine", "other", "needle"), + [ + ("self", None, None, "phone_vd.sh start"), + ("abc", None, None, "display id or 'self'"), + ("7", None, None, "not this session's"), + ( + "7", + _mine(9), + SessionDisplay("claude_2", 7, "sf", "x"), + "owned by claude_2", + ), + ], + ) + def test_refuses_what_is_not_this_sessions( + self, + capsys: pytest.CaptureFixture[str], + requested: str, + mine: SessionDisplay | None, + other: SessionDisplay | None, + needle: str, + ) -> None: + with ( + patch(f"{MOD}.own_display", return_value=mine), + patch(f"{MOD}.display_owner", return_value=other), + patch(f"{MOD}.AndroidUi") as factory, + patch(f"{MOD}.acquire") as acquire, + ): + assert cli.main(["--display", requested, "focus"]) == 5 + factory.assert_not_called() + acquire.assert_not_called() + assert needle in capsys.readouterr().err diff --git a/python_pkg/android_ui/tests/test_driver.py b/python_pkg/android_ui/tests/test_driver.py index c56e473e..fd9854dc 100644 --- a/python_pkg/android_ui/tests/test_driver.py +++ b/python_pkg/android_ui/tests/test_driver.py @@ -11,72 +11,18 @@ because each of them looked like success at the time: from __future__ import annotations -from pathlib import Path from typing import NamedTuple import pytest from python_pkg.android_ui import driver as drv +from python_pkg.android_ui.tests.conftest import FakeDevice +from python_pkg.android_ui.tests.conftest import node as _node +from python_pkg.android_ui.tests.conftest import tree as _tree MOD = "python_pkg.android_ui.driver" -def _node(text: str = "", **attrs: str) -> str: - """Render one ```` element for a fake dump. - - ``attrs`` accepts ``desc``, ``hint``, ``res``, ``cls``, ``bounds``, - ``enabled`` and ``focused``; each falls back to a representative default. - """ - return ( - f'' - ) - - -def _tree(*nodes: str) -> str: - """Wrap ``nodes`` in a hierarchy document.""" - return f"{''.join(nodes)}" - - -class FakeDevice: - """Scripted stand-in for adb, recording every command it is given.""" - - def __init__(self, trees: list[str] | None = None) -> None: - self.trees = trees or [_tree(_node("Connect"))] - self.calls: list[tuple[str, ...]] = [] - self.keyboard_shown = False - self.pulled = 0 - - def run(self, *args: str, timeout: float = 30.0) -> str: - """Answer an adb invocation.""" - del timeout - self.calls.append(args) - if args[:2] == ("shell", "dumpsys") and args[2] == "input_method": - return f"mInputShown={'true' if self.keyboard_shown else 'false'}" - if args[:2] == ("shell", "dumpsys") and args[2] == "window": - return "mCurrentFocus=Window{ab12 u0 com.example/com.example.Main}" - if args[0] == "pull": - index = min(self.pulled, len(self.trees) - 1) - self.pulled += 1 - Path(args[2]).write_text(self.trees[index], encoding="utf-8") - if args[:3] == ("shell", "input", "keyevent") and args[3] in {"111", "4"}: - self.keyboard_shown = False - return "" - - def taps(self) -> list[tuple[int, int]]: - """Return every tap coordinate, in order.""" - return [ - (int(c[3]), int(c[4])) - for c in self.calls - if c[:3] == ("shell", "input", "tap") - ] - - class Harness(NamedTuple): """A driver plus the fake device backing it. @@ -207,7 +153,9 @@ class TestTap: ui.device.trees = [_tree(_node("Go", bounds="[0,0][100,100]"))] ui.ui.tap("Go") keyevents = [ - c for c in ui.device.calls if c[:3] == ("shell", "input", "keyevent") + c + for c in ui.device.calls + if c[:5] == ("shell", "input", "-d", "0", "keyevent") ] assert keyevents, "expected the keyboard to be dismissed first" @@ -236,3 +184,26 @@ def test_real_text_outranks_the_hint_once_typed() -> None: (element,) = drv._parse_tree(_tree(dump)) assert element.label == "17:30" + + +class TestVirtualDisplay: + def test_reads_the_tree_through_the_helper_and_taps_with_d( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + device = FakeDevice() + instance = drv.AndroidUi(serial="S", display=9, settle_seconds=0.0) + monkeypatch.setattr(instance, "_run", device.run) + monkeypatch.setattr(f"{MOD}.time.sleep", lambda _s: None) + xml = _tree(_node("Go", bounds="[0,0][100,100]")) + calls: list[int] = [] + + def dump(run: object, display: int) -> str: + del run + calls.append(display) + return xml + + monkeypatch.setattr(f"{MOD}._a11y_helper.dump_display", dump) + instance.tap("Go") + assert calls + assert set(calls) == {9} + assert ("shell", "input", "-d", "9", "tap", "50", "50") in device.calls diff --git a/python_pkg/android_ui/tests/test_shell_helpers.py b/python_pkg/android_ui/tests/test_shell_helpers.py index c1012835..00fb50d7 100644 --- a/python_pkg/android_ui/tests/test_shell_helpers.py +++ b/python_pkg/android_ui/tests/test_shell_helpers.py @@ -54,16 +54,19 @@ class FakeDevice: index = min(self.pulled, len(self.trees) - 1) self.pulled += 1 Path(args[2]).write_text(self.trees[index], encoding="utf-8") - if args[:3] == ("shell", "input", "keyevent") and args[3] in {"111", "4"}: + if args[:5] == ("shell", "input", "-d", "0", "keyevent") and args[5] in { + "111", + "4", + }: self.keyboard_shown = False return "" def taps(self) -> list[tuple[int, int]]: """Return every tap coordinate, in order.""" return [ - (int(c[3]), int(c[4])) + (int(c[5]), int(c[6])) for c in self.calls - if c[:3] == ("shell", "input", "tap") + if c[:5] == ("shell", "input", "-d", "0", "tap") ] diff --git a/python_pkg/android_ui/tests/test_text_entry.py b/python_pkg/android_ui/tests/test_text_entry.py index 85944b65..c2f3db90 100644 --- a/python_pkg/android_ui/tests/test_text_entry.py +++ b/python_pkg/android_ui/tests/test_text_entry.py @@ -54,16 +54,19 @@ class FakeDevice: index = min(self.pulled, len(self.trees) - 1) self.pulled += 1 Path(args[2]).write_text(self.trees[index], encoding="utf-8") - if args[:3] == ("shell", "input", "keyevent") and args[3] in {"111", "4"}: + if args[:5] == ("shell", "input", "-d", "0", "keyevent") and args[5] in { + "111", + "4", + }: self.keyboard_shown = False return "" def taps(self) -> list[tuple[int, int]]: """Return every tap coordinate, in order.""" return [ - (int(c[3]), int(c[4])) + (int(c[5]), int(c[6])) for c in self.calls - if c[:3] == ("shell", "input", "tap") + if c[:5] == ("shell", "input", "-d", "0", "tap") ] @@ -98,7 +101,9 @@ class TestKeyboard: def test_dismiss_is_a_no_op_when_already_closed(self, ui: Harness) -> None: ui.ui.dismiss_keyboard() assert not [ - c for c in ui.device.calls if c[:3] == ("shell", "input", "keyevent") + c + for c in ui.device.calls + if c[:5] == ("shell", "input", "-d", "0", "keyevent") ] def test_closes_on_the_second_keyevent_when_escape_is_ignored( @@ -110,7 +115,9 @@ class TestKeyboard: monkeypatch.setattr(ui.ui, "keyboard_is_up", lambda: next(states, False)) ui.ui.dismiss_keyboard() sent = [ - c[3] for c in ui.device.calls if c[:3] == ("shell", "input", "keyevent") + c[5] + for c in ui.device.calls + if c[:5] == ("shell", "input", "-d", "0", "keyevent") ] assert sent == ["111", "4"] @@ -152,7 +159,7 @@ class TestTyping: deletes = [ c for c in ui.device.calls - if c[:3] == ("shell", "input", "keyevent") and c[3] == "67" + if c[:5] == ("shell", "input", "-d", "0", "keyevent") and c[5] == "67" ] assert deletes, "expected the field to be cleared first" diff --git a/python_pkg/phone_guard/__init__.py b/python_pkg/phone_guard/__init__.py new file mode 100644 index 00000000..4239f470 --- /dev/null +++ b/python_pkg/phone_guard/__init__.py @@ -0,0 +1,9 @@ +"""Keep several Claude sessions from interrupting each other on one phone. + +Two parts: :mod:`vd_state` answers "which virtual display is whose" from the +state ``phone_vd.sh`` keeps, and :mod:`hook` is the PreToolUse hook that stops +a raw ``adb`` command from touching the real screen (or another session's app) +without the matching :mod:`python_pkg.phone_lease` lease. +""" + +from __future__ import annotations diff --git a/python_pkg/phone_guard/__main__.py b/python_pkg/phone_guard/__main__.py new file mode 100644 index 00000000..f0ef9428 --- /dev/null +++ b/python_pkg/phone_guard/__main__.py @@ -0,0 +1,10 @@ +"""``python3 -m python_pkg.phone_guard``: the PreToolUse hook entry point.""" + +from __future__ import annotations + +import sys + +from python_pkg.phone_guard.hook import main + +if __name__ == "__main__": + sys.exit(main()) diff --git a/python_pkg/phone_guard/_need.py b/python_pkg/phone_guard/_need.py new file mode 100644 index 00000000..3b1e1df4 --- /dev/null +++ b/python_pkg/phone_guard/_need.py @@ -0,0 +1,18 @@ +"""The answer :mod:`classify` gives for one adb call.""" + +from __future__ import annotations + +from dataclasses import dataclass + +SCREEN = "screen" +WHOLE_PHONE = "whole-phone" + + +@dataclass(frozen=True) +class Need: + """What one adb call needs: a lease scope, a block, or nothing.""" + + scope: str | None = None # SCREEN, WHOLE_PHONE, or "pkg:" + block: str | None = None # why it may not run at all + display: int | None = None # the virtual display it targets, if any + serial: str | None = None # from ``-s`` diff --git a/python_pkg/phone_guard/_remote.py b/python_pkg/phone_guard/_remote.py new file mode 100644 index 00000000..7d939066 --- /dev/null +++ b/python_pkg/phone_guard/_remote.py @@ -0,0 +1,119 @@ +"""Per-tool rules for ``adb shell`` commands, for :mod:`classify`. + +Split out of ``classify.py`` for the 250-line cap. Each handler takes the +remote command's arguments (after the tool name) and returns the +:class:`~python_pkg.phone_guard.classify.Need` without a serial; the caller +stamps the serial on. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from python_pkg.phone_guard._need import SCREEN, WHOLE_PHONE, Need + +if TYPE_CHECKING: + from collections.abc import Callable + +_PM_PACKAGE = { + "clear", + "uninstall", + "disable", + "disable-user", + "enable", + "hide", + "unhide", + "suspend", + "unsuspend", + "grant", + "revoke", +} +_IDLE_WRITE = {"force-idle", "step", "unforce", "enable", "disable", "force-inactive"} +# `cmd connectivity set-package-networking-enabled ` +_NETWORKING_PKG_INDEX = 3 + + +def _display_flag(args: list[str], flag: str) -> int | None: + if flag in args: + i = args.index(flag) + if i + 1 < len(args) and args[i + 1].isdigit(): + return int(args[i + 1]) + return None + + +# `pm` options that take a value, so the value is not mistaken for the package. +_PM_VALUED = {"--user", "-u"} + + +def _package_after(args: list[str], verb: str) -> str | None: + rest = args[args.index(verb) + 1 :] + i = 0 + while i < len(rest): + if rest[i] in _PM_VALUED: + i += 2 + elif rest[i].startswith("-"): + i += 1 + else: + return rest[i] + return None + + +def _am(rest: list[str]) -> Need: + first = rest[0] if rest else "" + if first in {"force-stop", "kill", "stop-app"} and len(rest) > 1: + return Need(scope=f"pkg:{rest[-1]}") + if first in {"start", "start-activity", "startservice", "start-foreground-service"}: + display = _display_flag(rest, "--display") + return Need(display=display) if display else Need(scope=SCREEN) + return Need() + + +def _pm(rest: list[str]) -> Need: + if not rest or rest[0] not in _PM_PACKAGE: + return Need() + package = _package_after(rest, rest[0]) + return Need(scope=f"pkg:{package}" if package else SCREEN) + + +def _cmd(rest: list[str]) -> Need: + if rest[:2] != ["connectivity", "set-package-networking-enabled"]: + is_set = ( + rest[:1] == ["connectivity"] + and rest[1:2] != [] + and rest[1].startswith(("set-", "airplane-mode")) + ) + return Need(scope=WHOLE_PHONE) if is_set else Need() + package = rest[3] if len(rest) > _NETWORKING_PKG_INDEX else None + return Need(scope=f"pkg:{package}") if package else Need(scope=WHOLE_PHONE) + + +def _dumpsys(rest: list[str]) -> Need: + writes = (rest[:1] == ["deviceidle"] and rest[1:2] and rest[1] in _IDLE_WRITE) or ( + rest[:1] == ["battery"] and rest[1:2] and rest[1] in {"set", "unplug", "reset"} + ) + return Need(scope=WHOLE_PHONE) if writes else Need() + + +def _settings(rest: list[str]) -> Need: + return Need(scope=WHOLE_PHONE) if rest[:1] == ["put"] else Need() + + +def _wm(rest: list[str]) -> Need: + changes = rest[:1] in (["size"], ["density"]) and len(rest) > 1 + return Need(scope=SCREEN) if changes else Need() + + +REMOTE: dict[str, Callable[[list[str]], Need]] = { + "am": _am, + "pm": _pm, + "cmd": _cmd, + "dumpsys": _dumpsys, + "settings": _settings, + "wm": _wm, + "monkey": lambda _rest: Need(scope=SCREEN), + "svc": lambda _rest: Need(scope=WHOLE_PHONE), + "reboot": lambda _rest: Need(scope=WHOLE_PHONE), + "setprop": lambda _rest: Need(scope=WHOLE_PHONE), + "stop": lambda _rest: Need(scope=WHOLE_PHONE), + "start": lambda _rest: Need(scope=WHOLE_PHONE), +} diff --git a/python_pkg/phone_guard/classify.py b/python_pkg/phone_guard/classify.py new file mode 100644 index 00000000..8a976696 --- /dev/null +++ b/python_pkg/phone_guard/classify.py @@ -0,0 +1,165 @@ +"""Decide what a shell command's ``adb`` calls need before they may run. + +Pure: no adb, no filesystem. The hook feeds it the Bash tool's command and +applies the answer. Every ``adb`` invocation in the command gets one +:class:`Need`: + +* nothing -- reads (``logcat``, ``dumpsys``, ``getprop``, ``pull``, + ``screencap``), file transfer, and anything aimed at the caller's own + virtual display; +* a lease on the real screen -- input or launches on display 0; +* a lease on one app -- ``force-stop``, ``install``, per-app network cuts; +* the whole-phone lease -- reboot, Wi-Fi/airplane/Doze/battery tricks, + ``settings put``: everything that changes the phone for every session; +* a block -- key/text input without ``-d`` (it goes to whichever display last + took focus, i.e. possibly another session's app), or input aimed at another + session's virtual display. + +Unknown commands pass: this guards against interrupting other sessions, it is +not a sandbox. +""" + +from __future__ import annotations + +import re +import shlex + +from python_pkg.phone_guard._need import SCREEN, WHOLE_PHONE, Need +from python_pkg.phone_guard._remote import REMOTE, _display_flag + +_SEPARATORS = {";", "&&", "||", "|", "&", "(", ")", "\n"} +_GLOBAL_WITH_ARG = {"-s", "-t", "-H", "-P", "-L"} +_HEREDOC = re.compile(r"<<-?\s*(['\"]?)(\w+)\1[^\n]*\n.*?\n\s*\2\s*(?=\n|$)", re.DOTALL) + +_WHOLE_PHONE_ADB = { + "reboot", + "root", + "unroot", + "remount", + "disable-verity", + "enable-verity", + "tcpip", + "usb", + "kill-server", + "sideload", +} +_READ_ADB = { + "devices", + "get-serialno", + "get-state", + "version", + "help", + "logcat", + "pull", + "push", + "wait-for-device", + "forward", + "reverse", + "bugreport", + "start-server", + "sync", +} +_KEYS = {"text", "keyevent", "keycombination"} +_POINTER = {"tap", "swipe", "draganddrop", "press", "roll", "motionevent", "scroll"} + + +def _tokens(command: str) -> list[str]: + """Shell tokens, heredoc bodies removed (a script being WRITTEN is not run).""" + lexer = shlex.shlex( + _HEREDOC.sub("\n", command), posix=True, punctuation_chars=";&|()" + ) + lexer.whitespace = " \t\r" + lexer.whitespace_split = True + try: + return list(lexer) + except ValueError: + return [] + + +def _split(tokens: list[str]) -> list[list[str]]: + """Split a token stream into simple commands.""" + groups: list[list[str]] = [[]] + for word in tokens: + if word in _SEPARATORS or set(word) <= set(";&|"): + groups.append([]) + else: + groups[-1].append(word) + return [g for g in groups if g] + + +def adb_calls(command: str) -> list[list[str]]: + """Every ``adb ...`` simple command, as its argument list (no ``adb``).""" + calls = [] + for simple in _split(_tokens(command)): + for i, word in enumerate(simple): + if word == "adb" or word.endswith("/adb"): + calls.append(simple[i + 1 :]) + break + return calls + + +def _input(args: list[str], serial: str | None) -> Need: + display = _display_flag(args, "-d") + verb = next((a for a in args if a in _KEYS | _POINTER), "") + if display is not None and display != 0: + return Need(display=display, serial=serial) + if display is None and verb in _KEYS: + return Need( + block=f"`input {verb}` without -d goes to whichever display last took " + "focus -- possibly another session's app. Add `-d 0` for the real " + "screen or `-d ` (phone_vd.sh id).", + serial=serial, + ) + return Need(scope=SCREEN, serial=serial) + + +def _remote(args: list[str], serial: str | None) -> list[Need]: + """Needs of one ``adb shell`` command line.""" + found = [] + for simple in _split(_tokens(" ".join(args))): + tool, rest = simple[0], simple[1:] + found.append(_remote_one(tool, rest, serial)) + return found + + +def _remote_one(tool: str, rest: list[str], serial: str | None) -> Need: + """The need of one remote command, stamped with the call's serial.""" + if tool == "input": + return _input(rest, serial) + need = REMOTE.get(tool, lambda _rest: Need())(rest) + return Need(scope=need.scope, block=need.block, display=need.display, serial=serial) + + +def needs(command: str, apk_package: dict[str, str] | None = None) -> list[Need]: + """Every adb call's :class:`Need` in ``command``. + + ``apk_package`` maps an APK path to its package, resolved by the caller + (reading an APK is I/O); an unknown APK's install needs the screen. + """ + known = apk_package or {} + result: list[Need] = [] + for call in adb_calls(command): + serial = None + i = 0 + while i < len(call) and call[i].startswith("-"): + if call[i] in _GLOBAL_WITH_ARG and i + 1 < len(call): + serial = call[i + 1] if call[i] == "-s" else serial + i += 2 + else: + i += 1 + sub, args = (call[i], call[i + 1 :]) if i < len(call) else ("", []) + if sub in {"shell", "exec-out"}: + result.extend(_remote(args, serial)) + elif sub in _WHOLE_PHONE_ADB: + result.append(Need(scope=WHOLE_PHONE, serial=serial)) + elif sub in {"install", "install-multiple", "uninstall"}: + target = next((a for a in reversed(args) if not a.startswith("-")), "") + package = known.get(target) or (target if sub == "uninstall" else None) + result.append( + Need(scope=f"pkg:{package}" if package else SCREEN, serial=serial) + ) + elif sub in _READ_ADB or not sub: + result.append(Need(serial=serial)) + else: + result.append(Need(serial=serial)) + return result diff --git a/python_pkg/phone_guard/hook.py b/python_pkg/phone_guard/hook.py new file mode 100644 index 00000000..0aff7c8e --- /dev/null +++ b/python_pkg/phone_guard/hook.py @@ -0,0 +1,138 @@ +"""PreToolUse hook: lease -- or refuse -- raw ``adb`` before the Bash tool runs it. + +``android_ui``, ``phone_deploy.sh`` and ``phone_vd.sh`` take their own leases, +but a bare ``adb shell input tap`` in a Bash call took none, which is how one +session's tap landed in another session's app (2026-09-25). Claude Code pipes +the tool call here as JSON; exit 2 blocks it and shows stderr to the model. + +Leases are taken as this session (``phone_lease.owner_id`` finds the +``claude`` ancestor), so a later ``android_ui`` call from the same session +simply refreshes them. +""" + +from __future__ import annotations + +import json +import subprocess +import sys +from typing import TYPE_CHECKING + +from python_pkg.android_ui._a11y_helper import build_tool +from python_pkg.android_ui._elements import UiAutomationError +from python_pkg.phone_guard._need import SCREEN, WHOLE_PHONE +from python_pkg.phone_guard.classify import adb_calls, needs +from python_pkg.phone_guard.vd_state import display_owner, own_display +from python_pkg.phone_lease import ( + NoPhoneError, + PhoneBusyError, + Terms, + acquire, + owner_id, + resolve_serial, + scope_of, +) + +if TYPE_CHECKING: + from collections.abc import Iterable + from typing import TextIO + + from python_pkg.phone_guard._need import Need + +BLOCK = 2 +# Under Claude Code's 60 s hook timeout, with room to report. +_WAIT_SECONDS = 40.0 + + +def _apk_packages(command: str) -> dict[str, str]: + """Map each APK an ``adb install`` names to its package (via aapt2).""" + found: dict[str, str] = {} + try: + aapt2 = str(build_tool("aapt2")) + except UiAutomationError: + return found # no SDK: an unknown APK's install takes the screen lease + for call in adb_calls(command): + for word in call: + if not word.endswith(".apk"): + continue + try: + done = subprocess.run( + [aapt2, "dump", "packagename", word], + capture_output=True, + text=True, + timeout=10, + check=False, + ) + except OSError, subprocess.TimeoutExpired: + continue + if done.returncode == 0 and done.stdout.strip(): + found[word] = done.stdout.strip() + return found + + +def _scope(need: Need, serial: str, owner: str) -> str: + """The lease scope a need maps to; raises PermissionError for a foreign display.""" + if need.display is not None: + mine = own_display(serial, owner) + if mine is not None and mine.display == need.display: + return scope_of(mine.package) + other = display_owner(serial, need.display) + whose = f"session {other.owner_dir}'s" if other else "not your" + msg = ( + f"display {need.display} is {whose} virtual display; drive only " + "your own (`phone_vd.sh id`)" + ) + raise PermissionError(msg) + if need.scope == SCREEN: + return scope_of() + if need.scope == WHOLE_PHONE: + return scope_of(whole_phone=True) + return str(need.scope) + + +def decide(command: str, note: str = "") -> str | None: + """Take every lease ``command`` needs; return why it must not run, if so.""" + found: Iterable[Need] = needs(command, _apk_packages(command)) + owner = owner_id() + for need in found: + if need.block: + return need.block + if need.scope is None and need.display is None: + continue + try: + serial = resolve_serial(need.serial) + except NoPhoneError: + continue # no phone or several without -s: adb itself will refuse + try: + scope = _scope(need, serial, owner) + acquire( + serial, + note or f"hook: {command[:60]}", + scope=scope, + terms=Terms(wait=_WAIT_SECONDS), + owner=owner, + ) + except PermissionError as exc: + return str(exc) + except PhoneBusyError as exc: + return ( + f"{exc} -- wait for it, use your own virtual display, or ask the user" + ) + return None + + +def main(stdin: TextIO | None = None) -> int: + """Hook entry point: 0 lets the Bash call run, 2 blocks it.""" + try: + payload = json.load(stdin or sys.stdin) + except json.JSONDecodeError: + return 0 + if payload.get("tool_name") != "Bash": + return 0 + command = str((payload.get("tool_input") or {}).get("command") or "") + if "adb" not in command: + return 0 + reason = decide(command) + if reason is None: + return 0 + sys.stderr.write(f"phone_guard: {reason}\n") + return BLOCK diff --git a/python_pkg/phone_guard/tests/__init__.py b/python_pkg/phone_guard/tests/__init__.py new file mode 100644 index 00000000..4826e502 --- /dev/null +++ b/python_pkg/phone_guard/tests/__init__.py @@ -0,0 +1 @@ +"""Tests for :mod:`python_pkg.phone_guard`.""" diff --git a/python_pkg/phone_guard/tests/conftest.py b/python_pkg/phone_guard/tests/conftest.py new file mode 100644 index 00000000..114a5905 --- /dev/null +++ b/python_pkg/phone_guard/tests/conftest.py @@ -0,0 +1,33 @@ +"""Keep every phone_guard test off the live lease and virtual-display state.""" + +from __future__ import annotations + +from pathlib import Path +from typing import TYPE_CHECKING +from unittest.mock import patch + +import pytest + +from python_pkg.phone_lease import lease as lease_mod + +if TYPE_CHECKING: + from collections.abc import Iterator + + +@pytest.fixture(autouse=True) +def isolated_state(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Iterator[Path]: + """Redirect ~/.cache/phone-lease and /tmp/phone_vd into tmp_path.""" + monkeypatch.setenv("PHONE_VD_STATE_DIR", str(tmp_path / "phone_vd")) + with patch.object(lease_mod, "LEASE_DIR", tmp_path / "leases"): + yield tmp_path + + +def make_display( + root: Path, serial: str, owner_dir: str, display: str, package: str +) -> None: + """Write the state phone_vd.sh leaves for one running display.""" + state = root / "phone_vd" / serial / owner_dir + state.mkdir(parents=True) + (state / "display").write_text(display) + (state / "sf").write_text("123") + (state / "package").write_text(package) diff --git a/python_pkg/phone_guard/tests/test_classify.py b/python_pkg/phone_guard/tests/test_classify.py new file mode 100644 index 00000000..bdccc29d --- /dev/null +++ b/python_pkg/phone_guard/tests/test_classify.py @@ -0,0 +1,121 @@ +"""What each adb call needs, from the Bash command that runs it.""" + +from __future__ import annotations + +import pytest + +from python_pkg.phone_guard._need import SCREEN, WHOLE_PHONE, Need +from python_pkg.phone_guard.classify import adb_calls, needs + + +def only(command: str) -> Need: + found = needs(command) + assert len(found) == 1, found + return found[0] + + +@pytest.mark.parametrize( + ("command", "scope"), + [ + # reads and file transfer never need a lease + ("adb devices", None), + ("adb logcat -d", None), + ("adb pull /sdcard/a.png /tmp", None), + ("adb push a.jpg /sdcard/", None), + ("adb shell getprop ro.x", None), + ("adb exec-out screencap -p", None), + ("adb shell dumpsys window", None), + ("adb shell settings get global x", None), + ("adb shell cmd connectivity get-chain3-enabled", None), + ("adb shell cmd package list", None), + ("adb shell pm list packages", None), + ("adb shell am broadcast -a X", None), + ("adb shell am instrument -w a/.B", None), + ("adb shell wm size", None), + ("adb shell uiautomator dump", None), + ("adb", None), + ("adb frobnicate", None), + # the real screen + ("adb shell input tap 1 2", SCREEN), + ("adb shell input -d 0 text hi", SCREEN), + ("adb shell input swipe 1 2 3 4", SCREEN), + ("adb shell am start -n a/.B", SCREEN), + ("adb shell monkey -p a 1", SCREEN), + ("adb shell wm size 1080x2400", SCREEN), + ("adb shell pm clear --user 0", SCREEN), + ("adb install -r unknown.apk", SCREEN), + # one app + ("adb shell am force-stop dev.kuhy.todo", "pkg:dev.kuhy.todo"), + ("adb shell pm grant a.b android.permission.X", "pkg:a.b"), + ("adb shell pm clear --user 0 -f a.b", "pkg:a.b"), + ( + "adb shell cmd connectivity set-package-networking-enabled false a.b", + "pkg:a.b", + ), + ("adb uninstall a.b", "pkg:a.b"), + # the whole phone + ("adb reboot", WHOLE_PHONE), + ("adb kill-server", WHOLE_PHONE), + ("adb shell svc wifi disable", WHOLE_PHONE), + ("adb shell settings put global airplane_mode_on 1", WHOLE_PHONE), + ("adb shell cmd connectivity airplane-mode enable", WHOLE_PHONE), + ("adb shell cmd connectivity set-chain3-enabled true", WHOLE_PHONE), + ( + "adb shell cmd connectivity set-package-networking-enabled false", + WHOLE_PHONE, + ), + ("adb shell dumpsys deviceidle force-idle", WHOLE_PHONE), + ("adb shell dumpsys battery unplug", WHOLE_PHONE), + ("adb shell reboot", WHOLE_PHONE), + ("adb shell setprop a b", WHOLE_PHONE), + ], +) +def test_scopes(command: str, scope: str | None) -> None: + need = only(command) + assert need.scope == scope + assert need.block is None + + +def test_text_without_display_is_blocked() -> None: + for verb in ("text hi", "keyevent 4", "keycombination 113 29"): + assert "without -d" in (only(f"adb shell input {verb}").block or "") + + +def test_a_virtual_display_is_reported_not_leased() -> None: + need = only("adb shell input -d 9 text hi") + assert (need.display, need.scope) == (9, None) + assert only("adb shell am start --display 9 -n a/.B").display == 9 + + +def test_a_display_flag_without_a_number_is_not_a_display() -> None: + assert only("adb shell am start --display main -n a/.B").display is None + + +def test_serial_and_global_options() -> None: + need = only("adb -s R58M1 -t 3 shell am force-stop a.b") + assert (need.serial, need.scope) == ("R58M1", "pkg:a.b") + assert only("adb -e -H host shell input tap 1 1").scope == SCREEN + + +def test_installs_use_the_resolved_package() -> None: + found = needs("adb install -r -t app.apk", {"app.apk": "a.b"}) + assert found[0].scope == "pkg:a.b" + + +def test_every_call_in_a_compound_command() -> None: + found = needs( + "adb logcat -d | grep x && " + "/opt/sdk/adb -s S shell 'svc data disable; input tap 1 1'" + ) + assert [n.scope for n in found] == [None, WHOLE_PHONE, SCREEN] + + +def test_text_that_only_mentions_adb_is_ignored() -> None: + assert needs('echo "adb shell input tap 1 1"') == [] + assert needs("cat <<'EOF' > s.sh\nadb shell input tap 1 1\nEOF\nls") == [] + assert needs("grep adbd log") == [] + + +def test_unparseable_quoting_is_left_alone() -> None: + assert needs("adb shell 'input tap") == [] + assert adb_calls("") == [] diff --git a/python_pkg/phone_guard/tests/test_hook.py b/python_pkg/phone_guard/tests/test_hook.py new file mode 100644 index 00000000..b65ffdd1 --- /dev/null +++ b/python_pkg/phone_guard/tests/test_hook.py @@ -0,0 +1,128 @@ +"""The PreToolUse hook: which Bash calls run, which lease first, which stop.""" + +from __future__ import annotations + +import io +import json +from pathlib import Path +import subprocess +from unittest.mock import MagicMock, patch + +import pytest + +from python_pkg.android_ui._elements import UiAutomationError +from python_pkg.phone_guard import hook +from python_pkg.phone_guard.tests.conftest import make_display +from python_pkg.phone_lease import NoPhoneError, PhoneBusyError + +MOD = "python_pkg.phone_guard.hook" + + +def run(command: str, tool: str = "Bash") -> int: + return hook.main( + io.StringIO(json.dumps({"tool_name": tool, "tool_input": {"command": command}})) + ) + + +@pytest.fixture +def acquire() -> MagicMock: + with ( + patch(f"{MOD}.acquire") as mock, + patch(f"{MOD}.owner_id", return_value="claude:1"), + patch(f"{MOD}.resolve_serial", side_effect=lambda s=None: s or "S"), + patch(f"{MOD}.build_tool", side_effect=UiAutomationError("no sdk")), + ): + yield mock + + +class TestMain: + def test_ignores_what_it_cannot_or_need_not_judge(self, acquire: MagicMock) -> None: + assert hook.main(io.StringIO("not json")) == 0 + assert run("adb shell input tap 1 1", tool="Read") == 0 + assert run("ls -la") == 0 + assert hook.main(io.StringIO(json.dumps({"tool_name": "Bash"}))) == 0 + acquire.assert_not_called() + + def test_reads_pass_without_a_lease(self, acquire: MagicMock) -> None: + assert run("adb logcat -d") == 0 + acquire.assert_not_called() + + def test_a_block_exits_2_with_the_reason( + self, acquire: MagicMock, capsys: pytest.CaptureFixture[str] + ) -> None: + assert run("adb shell input text hi") == hook.BLOCK + assert "without -d" in capsys.readouterr().err + acquire.assert_not_called() + + +class TestLeases: + def test_the_real_screen(self, acquire: MagicMock) -> None: + assert run("adb shell input tap 1 1") == 0 + assert acquire.call_args.kwargs["scope"] == "pkg:__screen__" + assert acquire.call_args.kwargs["owner"] == "claude:1" + assert acquire.call_args.args[0] == "S" + + def test_the_whole_phone_and_an_app(self, acquire: MagicMock) -> None: + assert run("adb -s X reboot && adb shell am force-stop a.b") == 0 + scopes = [c.kwargs["scope"] for c in acquire.call_args_list] + assert scopes == ["whole-phone", "pkg:a.b"] + assert acquire.call_args_list[0].args[0] == "X" + + def test_own_display_takes_its_app_lease( + self, acquire: MagicMock, isolated_state: Path + ) -> None: + make_display(isolated_state, "S", "claude_1", "9", "dev.kuhy.todo") + assert run("adb shell input -d 9 tap 5 5") == 0 + assert acquire.call_args.kwargs["scope"] == "pkg:dev.kuhy.todo" + + def test_another_sessions_display_is_refused( + self, + acquire: MagicMock, + isolated_state: Path, + capsys: pytest.CaptureFixture[str], + ) -> None: + make_display(isolated_state, "S", "claude_2", "9", "x.y") + assert run("adb shell input -d 9 tap 5 5") == hook.BLOCK + assert "session claude_2's" in capsys.readouterr().err + assert run("adb shell input -d 4 tap 5 5") == hook.BLOCK + assert "not your" in capsys.readouterr().err + acquire.assert_not_called() + + def test_busy_phone_blocks_naming_the_holder( + self, acquire: MagicMock, capsys: pytest.CaptureFixture[str] + ) -> None: + acquire.side_effect = PhoneBusyError("held by claude:2") + assert run("adb shell input tap 1 1") == hook.BLOCK + err = capsys.readouterr().err + assert "held by claude:2" in err + assert "own virtual display" in err + + def test_no_resolvable_phone_is_left_to_adb(self, acquire: MagicMock) -> None: + with patch(f"{MOD}.resolve_serial", side_effect=NoPhoneError("several")): + assert run("adb shell input tap 1 1") == 0 + acquire.assert_not_called() + + +class TestApkPackages: + def test_reads_the_package_with_aapt2(self, tmp_path: Path) -> None: + done = subprocess.CompletedProcess([], 0, "a.b\n", "") + with ( + patch(f"{MOD}.build_tool", return_value=tmp_path / "aapt2"), + patch(f"{MOD}.subprocess.run", return_value=done) as runner, + ): + assert hook._apk_packages("adb install -r app.apk other.txt") == { + "app.apk": "a.b" + } + assert runner.call_args.args[0][0] == str(tmp_path / "aapt2") + + def test_failures_leave_the_apk_unknown(self, tmp_path: Path) -> None: + with patch(f"{MOD}.build_tool", return_value=tmp_path / "aapt2"): + with patch(f"{MOD}.subprocess.run", side_effect=OSError("gone")): + assert hook._apk_packages("adb install a.apk") == {} + failed = subprocess.CompletedProcess([], 1, "", "bad apk") + with patch(f"{MOD}.subprocess.run", return_value=failed): + assert hook._apk_packages("adb install a.apk") == {} + + def test_no_sdk_means_no_lookup(self) -> None: + with patch(f"{MOD}.build_tool", side_effect=UiAutomationError("no sdk")): + assert hook._apk_packages("adb install a.apk") == {} diff --git a/python_pkg/phone_guard/tests/test_main_entry.py b/python_pkg/phone_guard/tests/test_main_entry.py new file mode 100644 index 00000000..5819c49a --- /dev/null +++ b/python_pkg/phone_guard/tests/test_main_entry.py @@ -0,0 +1,17 @@ +"""``python -m python_pkg.phone_guard`` runs the hook.""" + +from __future__ import annotations + +import runpy +from unittest.mock import patch + +import pytest + + +def test_module_entry_exits_with_the_hook_code() -> None: + with ( + patch("python_pkg.phone_guard.hook.main", return_value=2), + pytest.raises(SystemExit) as done, + ): + runpy.run_module("python_pkg.phone_guard", run_name="__main__") + assert done.value.code == 2 diff --git a/python_pkg/phone_guard/tests/test_vd_state.py b/python_pkg/phone_guard/tests/test_vd_state.py new file mode 100644 index 00000000..62d22064 --- /dev/null +++ b/python_pkg/phone_guard/tests/test_vd_state.py @@ -0,0 +1,45 @@ +"""Reading phone_vd.sh's per-session state.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from python_pkg.phone_guard import vd_state +from python_pkg.phone_guard.tests.conftest import make_display + + +def test_default_root_is_under_the_temp_dir(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("PHONE_VD_STATE_DIR") + assert vd_state.state_root().name == "phone_vd" + + +def test_owner_dir_name_matches_the_script() -> None: + assert vd_state.owner_dir_name("claude:123") == "claude_123" + + +def test_lists_and_finds_displays(isolated_state: Path) -> None: + make_display(isolated_state, "S", "claude_1", "9", "a.b") + make_display(isolated_state, "S", "claude_2", "10", "c.d") + make_display(isolated_state, "S", "claude_3", "gone", "e.f") + assert [d.display for d in vd_state.displays("S")] == [9, 10] + mine = vd_state.own_display("S", "claude:1") + assert mine is not None + assert (mine.display, mine.package, mine.sf) == (9, "a.b", "123") + assert vd_state.own_display("S", "claude:9") is None + other = vd_state.display_owner("S", 10) + assert other is not None + assert other.owner_dir == "claude_2" + assert vd_state.display_owner("S", 11) is None + + +def test_no_state_means_no_displays() -> None: + assert vd_state.displays("NOPE") == [] + + +def test_unreadable_files_read_as_empty(isolated_state: Path) -> None: + make_display(isolated_state, "S", "claude_1", "9", "a.b") + (isolated_state / "phone_vd" / "S" / "claude_1" / "package").unlink() + (isolated_state / "phone_vd" / "S" / "claude_1" / "package").mkdir() + assert vd_state.displays("S")[0].package == "" diff --git a/python_pkg/phone_guard/vd_state.py b/python_pkg/phone_guard/vd_state.py new file mode 100644 index 00000000..90885dd6 --- /dev/null +++ b/python_pkg/phone_guard/vd_state.py @@ -0,0 +1,83 @@ +"""Which session owns which virtual display, read from ``phone_vd.sh`` state. + +``phone_vd.sh`` keeps one directory per session under +``$PHONE_VD_STATE_DIR///`` holding ``display`` (the logical id +``input -d`` takes), ``sf`` (the SurfaceFlinger id ``screencap -d`` takes) and +``package``. The owner directory name is the phone_lease owner id with +anything outside ``[A-Za-z0-9_.-]`` replaced by ``_``, exactly as the script +writes it. +""" + +from __future__ import annotations + +from dataclasses import dataclass +import os +from pathlib import Path +import re +import tempfile + +_UNSAFE = re.compile(r"[^A-Za-z0-9_.-]") + + +def state_root() -> Path: + """The directory ``phone_vd.sh`` keeps its per-phone state in. + + ``$PHONE_VD_STATE_DIR``, else ``phone_vd`` under the temp dir -- the same + ``${TMPDIR:-/tmp}`` rule the script applies, so both sides agree. + """ + default = Path(tempfile.gettempdir()) / "phone_vd" + return Path(os.environ.get("PHONE_VD_STATE_DIR", str(default))) + + +def owner_dir_name(owner: str) -> str: + """``owner`` as ``phone_vd.sh`` names its state directory.""" + return _UNSAFE.sub("_", owner) + + +@dataclass(frozen=True) +class SessionDisplay: + """One running virtual display and whose it is.""" + + owner_dir: str + display: int + sf: str + package: str + + +def _read(path: Path) -> str: + try: + return path.read_text(encoding="utf-8").strip() + except OSError: + return "" + + +def displays(serial: str) -> list[SessionDisplay]: + """Every session's virtual display on ``serial``.""" + root = state_root() / serial + found: list[SessionDisplay] = [] + if not root.is_dir(): + return found + for entry in sorted(root.iterdir()): + display = _read(entry / "display") + if not display.isdigit(): + continue + found.append( + SessionDisplay( + owner_dir=entry.name, + display=int(display), + sf=_read(entry / "sf"), + package=_read(entry / "package"), + ) + ) + return found + + +def own_display(serial: str, owner: str) -> SessionDisplay | None: + """``owner``'s virtual display on ``serial``, if it has one running.""" + mine = owner_dir_name(owner) + return next((d for d in displays(serial) if d.owner_dir == mine), None) + + +def display_owner(serial: str, display: int) -> SessionDisplay | None: + """The session whose virtual display has logical id ``display``.""" + return next((d for d in displays(serial) if d.display == display), None)