tools: reuse paired Google ADB transports

2026-11-07 10:46:07 -10:00 · 2026-10-26 08:13:11 +01:00
1 parent baadd07a7d
commit 5834c199de
@@ -485,7 +485,7 @@ tools/theme.sh themes/aurora                                 # push a theme; --r

 | Tool | What it does |
 |---|---|
-| [`tools/tv.py`](tools/tv.py) | Pure-Python ADB client: shell, batch, push, pull, reboot, and screenshots, optionally pressing remote keys first. Works where the stock `adb` binary is blocked. |
+| [`tools/tv.py`](tools/tv.py) | Pure-Python ADB client: shell, batch, push, pull, reboot, and screenshots, optionally pressing remote keys first. Works where the stock `adb` binary is blocked. Optional [paired Google ADB backend](tools/native-adb.md) for encrypted Wireless debugging. |
 | [`tools/b64push.py`](tools/b64push.py) | Uploads large files such as APKs in base64 chunks and verifies the checksum. |
 | [`tools/relay.py`](tools/relay.py) | Forwards a local port to the TV's ADB port, for assistants whose sandbox can't reach your network. Listens on this computer only. [Guide](tools/relay.md). |
 | [`tools/disable.sh`](tools/disable.sh), [`restore.sh`](tools/restore.sh) | Apply or undo a debloat list, the undo from the state recorded before the change. |
@@ -0,0 +1,152 @@
+"""The opt-in Google ADB backend must keep every operation on the selected transport."""
+import importlib.util
+import contextlib
+import io
+import json
+import os
+from pathlib import Path
+import subprocess
+import sys
+import tempfile
+import types
+import unittest
+from unittest import mock
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+@unittest.skipIf(os.name == "nt", "fake adb executable uses a POSIX shebang")
+class NativeAdbTest(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory(prefix="native adb ")
+        self.addCleanup(self.tmp.cleanup)
+        self.dir = Path(self.tmp.name)
+        self.log = self.dir / "calls.jsonl"
+        self.binary = self.dir / "fake adb"
+        self.binary.write_text(f"#!{sys.executable}\n" + r'''
+import json, os, pathlib, shutil, subprocess, sys
+args = sys.argv[1:]
+with open(os.environ['FAKE_ADB_LOG'], 'a') as f:
+    f.write(json.dumps(args) + '\n')
+if os.environ.get('FAKE_ADB_FAIL'):
+    print('transport unavailable', file=sys.stderr)
+    sys.exit(1)
+operation = args[2]
+if operation == 'get-state':
+    print(os.environ.get('FAKE_ADB_STATE', 'device'))
+elif operation == 'shell':
+    if os.environ.get('FAKE_ADB_RUN_SHELL'):
+        result = subprocess.run(args[-1], shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+        sys.stdout.buffer.write(result.stdout)
+    else:
+        print('remote output\nremote stderr')
+elif operation == 'pull':
+    shutil.copyfile(os.environ['FAKE_ADB_REMOTE_FILE'], args[4])
+''')
+        self.binary.chmod(0o755)
+        self.environment = mock.patch.dict(os.environ, {
+            "TV_HOST": "192.0.2.8:37123", "TV_ADB_BINARY": str(self.binary),
+            "FAKE_ADB_LOG": str(self.log),
+        })
+        self.environment.start()
+        self.addCleanup(self.environment.stop)
+        spec = importlib.util.spec_from_file_location("tv_native_test", ROOT / "tools" / "tv.py")
+        self.tv = importlib.util.module_from_spec(spec)
+        # Native commands do not use Python ADB keys or sockets. Stub those imports so
+        # this test also runs under CI's system Python, which has no adb_shell installed.
+        stubs = {name: types.ModuleType(name) for name in (
+            "adb_shell", "adb_shell.auth", "adb_shell.adb_device",
+            "adb_shell.auth.keygen", "adb_shell.auth.sign_pythonrsa")}
+        stubs["adb_shell.adb_device"].AdbDeviceTcp = object
+        stubs["adb_shell.auth.keygen"].keygen = lambda path: None
+        stubs["adb_shell.auth.sign_pythonrsa"].PythonRSASigner = lambda pub, priv: None
+        with mock.patch.dict(sys.modules, stubs):
+            spec.loader.exec_module(self.tv)
+
+    def calls(self):
+        return [json.loads(line) for line in self.log.read_text().splitlines()]
+
+    def test_shell_does_not_execute_local_shell_syntax_and_keeps_remote_stderr(self):
+        marker = self.dir / "must-not-exist"
+        command = f"printf hello; touch '{marker}'"
+        with mock.patch.object(self.tv, "get_signer", side_effect=AssertionError("no key read")):
+            device = self.tv.connect(timeout=2)
+        self.assertEqual(device.shell(command), "remote output\nremote stderr\n")
+        device.close()
+        self.assertFalse(marker.exists())
+        self.assertEqual(self.calls(), [
+            ["-s", "192.0.2.8:37123", "get-state"],
+            ["-s", "192.0.2.8:37123", "shell", "-T", "-x", command],
+        ])
+
+    def test_binary_pull_push_and_reboot_stay_on_the_same_transport(self):
+        remote = self.dir / "remote bytes"
+        remote.write_bytes(bytes(range(256)))
+        local = self.dir / "pulled bytes"
+        with mock.patch.dict(os.environ, {"FAKE_ADB_REMOTE_FILE": str(remote)}):
+            device = self.tv.connect()
+            device.pull("/data/local/tmp/picture.png", local, read_timeout_s=3)
+            device.push(local, "/data/local/tmp/file with spaces", read_timeout_s=3)
+            device.reboot()
+            device.close()
+        self.assertEqual(local.read_bytes(), remote.read_bytes())
+        calls = self.calls()
+        self.assertTrue(all(c[:2] == ["-s", "192.0.2.8:37123"] for c in calls))
+        self.assertEqual([c[2] for c in calls], ["get-state", "pull", "push", "reboot"])
+        self.assertEqual(calls[2][-1], "/data/local/tmp/file with spaces")
+
+    def test_installer_base64_uploader_uses_native_transport_and_verifies_bytes(self):
+        # The installer uses b64push, not Device.push. Run its real chunking, shell
+        # checksums and rename through the fake ADB executable against temporary files.
+        with tempfile.TemporaryDirectory(prefix="native-b64-") as folder:
+            source = Path(folder) / "source.apk"
+            target = Path(folder) / "target.apk"
+            data = bytes(range(256)) * 40
+            source.write_bytes(data)
+            target.write_bytes(b"old content")
+            spec = importlib.util.spec_from_file_location("native_b64_test", ROOT / "tools" / "b64push.py")
+            uploader = importlib.util.module_from_spec(spec)
+            spec.loader.exec_module(uploader)
+            with mock.patch.dict(sys.modules, {"tv": self.tv}), \
+                    mock.patch.dict(os.environ, {"FAKE_ADB_RUN_SHELL": "1"}), \
+                    mock.patch.object(sys, "argv", ["b64push.py", str(source), str(target)]), \
+                    mock.patch.object(sys, "path", list(sys.path)), \
+                    contextlib.redirect_stdout(io.StringIO()) as output:
+                self.assertEqual(uploader.main(), 0)
+            self.assertEqual(target.read_bytes(), data)
+            self.assertIn("MATCH", output.getvalue())
+            self.assertFalse(Path(str(target) + ".part").exists())
+            chunks = [c for c in self.calls() if c[2] == "shell" and "&& echo ok" in c[-1]]
+            self.assertGreater(len(chunks), 1)
+            self.assertTrue(all(len(c[-1]) < 8192 for c in chunks))
+
+    def test_unauthorized_or_failed_transport_never_falls_back(self):
+        for env in ({"FAKE_ADB_STATE": "unauthorized"}, {"FAKE_ADB_FAIL": "1"}):
+            with self.subTest(env=env), mock.patch.dict(os.environ, env), \
+                    mock.patch.object(self.tv, "AdbDeviceTcp", side_effect=AssertionError("no fallback")):
+                with self.assertRaises(self.tv.NativeAdbError):
+                    self.tv.connect(timeout=2)
+        self.assertEqual([c[2] for c in self.calls()], ["get-state", "get-state"])
+
+    def test_subprocess_timeout_is_bounded_and_reported(self):
+        device = self.tv.connect(timeout=7)
+        with mock.patch.object(self.tv.subprocess, "run",
+                               side_effect=subprocess.TimeoutExpired("adb", 3)) as run:
+            with self.assertRaisesRegex(self.tv.NativeAdbError, "timed out after 3s"):
+                device.shell("getprop ro.product.model", read_timeout_s=3)
+        self.assertEqual(run.call_args.kwargs["timeout"], 3)
+
+    def test_ipv6_endpoint_is_explicit_and_bracketed(self):
+        device = self.tv.NativeAdbDevice(str(self.binary), "2001:db8::8", 37123, 2)
+        device.shell("getprop ro.product.model")
+        self.assertTrue(all(c[:2] == ["-s", "[2001:db8::8]:37123"] for c in self.calls()))
+
+    def test_missing_executable_is_a_clear_connection_error(self):
+        with mock.patch.dict(os.environ, {"TV_ADB_BINARY": str(self.dir / "absent")}):
+            with self.assertRaises(self.tv.NativeAdbError) as raised:
+                self.tv.connect(timeout=2)
+        self.assertIn("Could not run Google ADB", self.tv.connect_error(raised.exception))
+
+
+if __name__ == "__main__":
+    unittest.main()
@@ -0,0 +1,41 @@
+# Use a paired Google ADB connection
+
+The Python client remains the default. If the TV offers encrypted Wireless debugging,
+`adb_shell` cannot speak its TLS protocol. The optional Google ADB backend lets the existing
+tools use that connection without opening a plaintext ADB port.
+
+Install Google's platform-tools and the usual `tools/requirements.txt` dependencies first.
+On the TV, enable Wireless debugging. Pair with the address, port and code from **Pair device
+with pairing code**, then connect to the separate address and port on the main Wireless
+debugging screen:
+
+```bash
+adb pair <tv-ip>:<pairing-port>
+adb connect <tv-ip>:<connection-port>
+export TV_ADB_BINARY=/absolute/path/to/adb
+export TV_HOST=<tv-ip>:<connection-port>
+tools/venv/bin/python -I tools/tv.py --probe
+```
+
+Google ADB manages pairing and authentication; the tools neither read its private key nor
+implement TLS. Keep the main screen's current connection port: it can change when debugging
+is toggled or the TV restarts. `TV_HOST` and `TV_PORT` have their usual meaning. IPv6 endpoints
+use `[address]:port`. The executable may also be a name on PATH, such as `adb`.
+
+The backend requires an already connected, authorized transport. Every operation selects
+that exact endpoint with `adb -s`; it never falls back to another TV or runs `adb tcpip`
+to change the TV's debugging mode. Its `close()` leaves the shared Google ADB connection
+and server running, so other tools using them keep working.
+
+The setting is inherited by `tools/disable.sh`, `tools/restore.sh`, `tools/privacy.py`,
+`tools/theme.sh` and `launcher/build-and-install.sh`; their existing safety checks still
+run. Back up first and follow `PROMPT.md` before changing a TV. For example:
+
+```bash
+tools/venv/bin/python -I tools/tv.py --screenshot shots/before.png
+```
+
+When finished, switch Wireless debugging and any other enabled debugging modes off on the
+TV and revoke the authorizations. Check the noted connection port with
+`python3 tools/adb-closed.py <tv-ip> <connection-port>` before rebooting, then check that
+debugging remains off after the reboot. Pairing codes, keys and screenshots stay private.
@@ -11,9 +11,13 @@ Set TV_HOST to the TV's IP (or IP:port). Usage:
    tv.py --pull <remote> <local>
    tv.py --reboot                     return once the reboot request is accepted; the TV may still be restarting
    tv.py --help                       this text; needs neither the venv nor a TV
+
+Set TV_ADB_BINARY to the Google adb executable to use an already connected, paired transport
+(including encrypted Wireless debugging). See tools/native-adb.md.
 """
 import os
 import re
+import subprocess
 import sys
 import time

@@ -159,7 +163,67 @@ def stock_signer():
        return None


+class NativeAdbError(RuntimeError):
+    """The selected Google ADB transport is unavailable or an operation failed."""
+
+
+class NativeAdbDevice:
+    """The subset of adb_shell used by the tools, over an existing Google ADB transport.
+
+    Every command selects the same explicit endpoint. Never select the first connected TV,
+    change adbd's mode, or disconnect a transport another program may be using.
+    """
+
+    # b64push sizes its shell commands from this payload bound. A small value also
+    # keeps adb's command-line arguments below Windows' process argument limit.
+    max_chunk_size = 4096
+
+    def __init__(self, binary, host, port, timeout):
+        self.binary = binary
+        self.serial = f"[{host}]:{port}" if ":" in host else f"{host}:{port}"
+        self.timeout = float(timeout)
+        if self._run(["get-state"]).strip() != "device":
+            raise NativeAdbError(f"{self.serial} is not an authorized ADB transport; pair and connect it first.")
+
+    def _run(self, args, timeout=None):
+        limit = self.timeout if timeout is None else float(timeout)
+        try:
+            result = subprocess.run([self.binary, "-s", self.serial, *args],
+                                    capture_output=True, text=True, errors="replace", timeout=limit)
+        except subprocess.TimeoutExpired as e:
+            raise NativeAdbError(f"Google ADB {args[0]} on {self.serial} timed out after {limit:g}s.") from e
+        except OSError as e:
+            raise NativeAdbError(f"Could not run Google ADB {self.binary!r}: {e}") from e
+        if result.returncode:
+            detail = result.stderr.strip() or result.stdout.strip() or f"exit {result.returncode}"
+            raise NativeAdbError(f"Google ADB {args[0]} on {self.serial}: {detail}")
+        return result.stdout
+
+    def shell(self, command, read_timeout_s=None, **kwargs):
+        # -T avoids a PTY; -x matches adb_shell's shell service: remote stderr is in the
+        # returned text and a remote exit code is not mistaken for a transport failure.
+        # Guards still see remote Error/Failure lines. No local shell interprets command.
+        return self._run(["shell", "-T", "-x", command], read_timeout_s)
+
+    def push(self, local, remote, transport_timeout_s=None, read_timeout_s=None, **kwargs):
+        self._run(["push", os.fspath(local), os.fspath(remote)],
+                  read_timeout_s if read_timeout_s is not None else transport_timeout_s)
+
+    def pull(self, remote, local, transport_timeout_s=None, read_timeout_s=None, **kwargs):
+        self._run(["pull", os.fspath(remote), os.fspath(local)],
+                  read_timeout_s if read_timeout_s is not None else transport_timeout_s)
+
+    def reboot(self):
+        self._run(["reboot"])
+
+    def close(self):
+        pass  # Each subprocess is closed; leave the shared ADB server and transport alone.
+
+
 def connect(timeout=120):
+    binary = os.environ.get("TV_ADB_BINARY")
+    if binary:
+        return NativeAdbDevice(binary, HOST, PORT, timeout)
    dev = AdbDeviceTcp(HOST, PORT, default_transport_timeout_s=float(timeout))
    keys = [get_signer()]   # first, so the prompt asks for the same key as before
    stock = stock_signer()
@@ -176,6 +240,8 @@ def connect_error(e):
    about the port to use instead (#1578). connect() itself still raises: check.py (through
    recovery.connect()) and privacy.py catch its errors with wording and exit codes of their own, and
    recovery.py --save still lets them through (a follow-up)."""
+    if isinstance(e, NativeAdbError):
+        return str(e)
    # Imported here rather than with the client above, so the stub adb_shell in the tests (adb_device
    # and auth only) still loads this file.
    from adb_shell.exceptions import AdbTimeoutError, InvalidCommandError, TcpTimeoutException