blob: f36913293cd6e73fd872ce26f67d4e94a8d7a3fa [file] [log] [blame]
#!/usr/bin/env python3
# Copyright 2022 The Fuchsia Authors
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# See the License for the specific language governing permissions and
# limitations under the License.
import logging
import re
import shlex
import shutil
from antlion.controllers.adb_lib.error import AdbCommandError, AdbError
from antlion.libs.proc import job
ADB_REGEX = re.compile("adb:")
# Uses a regex to be backwards compatible with previous versions of ADB
# (N and above add the serial to the error msg).
DEVICE_NOT_FOUND_REGEX = re.compile("error: device (?:'.*?' )?not found")
DEVICE_OFFLINE_REGEX = re.compile("error: device offline")
# Raised when adb forward commands fail to forward a port.
CANNOT_BIND_LISTENER_REGEX = re.compile("error: cannot bind listener:")
# Expected output is "Android Debug Bridge version 1.0.XX
ADB_VERSION_REGEX = re.compile("Android Debug Bridge version 1.0.(\d+)")
GREP_REGEX = re.compile("grep(\s+)")
SHELL_USER_ID = "2000"
def parsing_parcel_output(output):
"""Parsing the adb output in Parcel format.
Parsing the adb output in format:
Result: Parcel(
0x00000000: 00000000 00000014 00390038 00340031 '........'
0x00000010: 00300038 00300030 00300030 00340032 ''
0x00000020: 00350034 00330035 00320038 00310033 ''
0x00000030: 00000000 '.... ')
output = "".join(re.findall(r"'(.*)'", output))
return re.sub(r"[.\s]", "", output)
class AdbProxy(object):
"""Proxy class for ADB.
For syntactic reasons, the '-' in adb commands need to be replaced with
'_'. Can directly execute adb commands on an object:
>> adb = AdbProxy(<serial>)
>> adb.start_server()
>> adb.devices() # will return the console output of "adb devices".
def __init__(self, serial="", ssh_connection=None):
"""Construct an instance of AdbProxy.
serial: str serial number of Android device from `adb devices`
ssh_connection: SshConnection instance if the Android device is
connected to a remote host that we can reach via SSH.
self.serial = serial
self._server_local_port = None
adb_path = shutil.which("adb")
adb_cmd = [shlex.quote(adb_path)]
if serial:
adb_cmd.append(f"-s {serial}")
if ssh_connection is not None:
# Kill all existing adb processes on the remote host (if any)
# Note that if there are none, then pkill exits with non-zero status"pkill adb", ignore_status=True)
# Copy over the adb binary to a temp dir
temp_dir ="mktemp -d").stdout.strip()
ssh_connection.send_file(adb_path, temp_dir)
# Start up a new adb server running as root from the copied binary.
remote_adb_cmd = "%s/adb %s root" % (
"-s %s" % serial if serial else "",
# Proxy a local port to the adb server port
local_port = ssh_connection.create_ssh_tunnel(5037)
self._server_local_port = local_port
if self._server_local_port:
adb_cmd.append(f"-P {local_port}")
self.adb_str = " ".join(adb_cmd)
self._ssh_connection = ssh_connection
def get_user_id(self):
"""Returns the adb user. Either 2000 (shell) or 0 (root)."""
return"id -u")
def is_root(self, user_id=None):
"""Checks if the user is root.
user_id: if supplied, the id to check against.
True if the user is root. False otherwise.
if not user_id:
user_id = self.get_user_id()
return user_id == ROOT_USER_ID
def ensure_root(self):
"""Ensures the user is root after making this call.
Note that this will still fail if the device is a user build, as root
is not accessible from a user build.
False if the device is a user build. True otherwise.
return self.is_root()
def ensure_user(self, user_id=SHELL_USER_ID):
"""Ensures the user is set to the given user.
user_id: The id of the user.
if self.is_root(user_id):
return self.get_user_id() == user_id
def _exec_cmd(self, cmd, ignore_status=False, timeout=DEFAULT_ADB_TIMEOUT):
"""Executes adb commands in a new shell.
This is specific to executing adb commands.
cmd: A string or list that is the adb command to execute.
The stdout of the adb command.
AdbError for errors in ADB operations.
AdbCommandError for errors from commands executed through ADB.
if isinstance(cmd, list):
cmd = " ".join(cmd)
result =, ignore_status=True, timeout=timeout)
ret, out, err = result.exit_status, result.stdout, result.stderr
if any(
for pattern in [
raise AdbError(cmd=cmd, stdout=out, stderr=err, ret_code=ret)
if "Result: Parcel" in out:
return parsing_parcel_output(out)
if ignore_status or (ret == 1 and
return out or err
if ret != 0:
raise AdbCommandError(cmd=cmd, stdout=out, stderr=err, ret_code=ret)
return out
def _exec_adb_cmd(self, name, arg_str, **kwargs):
return self._exec_cmd(" ".join((self.adb_str, name, arg_str)), **kwargs)
def _exec_cmd_nb(self, cmd, **kwargs):
"""Executes adb commands in a new shell, non blocking.
cmds: A string that is the adb command to execute.
return job.run_async(cmd, **kwargs)
def _exec_adb_cmd_nb(self, name, arg_str, **kwargs):
return self._exec_cmd_nb(" ".join((self.adb_str, name, arg_str)), **kwargs)
def tcp_forward(self, host_port, device_port):
"""Starts tcp forwarding from localhost to this android device.
host_port: Port number to use on localhost
device_port: Port number to use on the android device.
Forwarded port on host as int or command output string on error
if self._ssh_connection:
# We have to hop through a remote host first.
# 1) Find some free port on the remote host's localhost
# 2) Setup forwarding between that remote port and the requested
# device port
remote_port = self._ssh_connection.find_free_port()
host_port = self._ssh_connection.create_ssh_tunnel(
remote_port, local_port=host_port
output = self.forward(f"tcp:{host_port} tcp:{device_port}", ignore_status=True)
# If hinted_port is 0, the output will be the selected port.
# Otherwise, there will be no output upon successfully
# forwarding the hinted port.
if not output:
return host_port
output_int = int(output)
except ValueError:
return output
return output_int
def remove_tcp_forward(self, host_port):
"""Stop tcp forwarding a port from localhost to this android device.
host_port: Port number to use on localhost
if self._ssh_connection:
remote_port = self._ssh_connection.close_ssh_tunnel(host_port)
if remote_port is None:
"Cannot close unknown forwarded tcp port: %d", host_port
# The actual port we need to disable via adb is on the remote host.
host_port = remote_port
self.forward(f"--remove tcp:{host_port}")
def getprop(self, prop_name):
"""Get a property of the device.
This is a convenience wrapper for "adb shell getprop xxx".
prop_name: A string that is the name of the property to get.
A string that is the value of the property, or None if the property
doesn't exist.
return"getprop {prop_name}")
# TODO: This should be abstracted out into an object like the other shell
# command.
def shell(self, command, ignore_status=False, timeout=DEFAULT_ADB_TIMEOUT):
return self._exec_adb_cmd(
"shell", shlex.quote(command), ignore_status=ignore_status, timeout=timeout
def shell_nb(self, command):
return self._exec_adb_cmd_nb("shell", shlex.quote(command))
def __getattr__(self, name):
def adb_call(*args, **kwargs):
clean_name = name.replace("_", "-")
if clean_name in ["pull", "push", "remount"] and "timeout" not in kwargs:
kwargs["timeout"] = DEFAULT_ADB_PULL_TIMEOUT
arg_str = " ".join(str(elem) for elem in args)
return self._exec_adb_cmd(clean_name, arg_str, **kwargs)
return adb_call
def get_version_number(self):
"""Returns the version number of ADB as an int (XX in 1.0.XX).
AdbError if the version number is not found/parsable.
version_output = self.version()
match =, version_output)
if not match:
"Unable to capture ADB version from adb version "
"output: %s" % version_output
raise AdbError("adb version", version_output, "", "")
return int(