1#!/usr/bin/env python3 2# 3# Copyright 2016 - The Android Open Source Project 4# 5# Licensed under the Apache License, Version 2.0 (the "License"); 6# you may not use this file except in compliance with the License. 7# You may obtain a copy of the License at 8# 9# http://www.apache.org/licenses/LICENSE-2.0 10# 11# Unless required by applicable law or agreed to in writing, software 12# distributed under the License is distributed on an "AS IS" BASIS, 13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 14# See the License for the specific language governing permissions and 15# limitations under the License. 16 17from builtins import str 18 19import logging 20import re 21import shlex 22import shutil 23 24from acts.controllers.adb_lib.error import AdbCommandError 25from acts.controllers.adb_lib.error import AdbError 26from acts.libs.proc import job 27from acts.metrics.loggers import usage_metadata_logger 28 29DEFAULT_ADB_TIMEOUT = 60 30DEFAULT_ADB_PULL_TIMEOUT = 180 31 32ADB_REGEX = re.compile('adb:') 33# Uses a regex to be backwards compatible with previous versions of ADB 34# (N and above add the serial to the error msg). 35DEVICE_NOT_FOUND_REGEX = re.compile('error: device (?:\'.*?\' )?not found') 36DEVICE_OFFLINE_REGEX = re.compile('error: device offline') 37# Raised when adb forward commands fail to forward a port. 38CANNOT_BIND_LISTENER_REGEX = re.compile('error: cannot bind listener:') 39# Expected output is "Android Debug Bridge version 1.0.XX 40ADB_VERSION_REGEX = re.compile('Android Debug Bridge version 1.0.(\d+)') 41GREP_REGEX = re.compile('grep(\s+)') 42 43ROOT_USER_ID = '0' 44SHELL_USER_ID = '2000' 45 46 47def parsing_parcel_output(output): 48 """Parsing the adb output in Parcel format. 49 50 Parsing the adb output in format: 51 Result: Parcel( 52 0x00000000: 00000000 00000014 00390038 00340031 '........8.9.1.4.' 53 0x00000010: 00300038 00300030 00300030 00340032 '8.0.0.0.0.0.2.4.' 54 0x00000020: 00350034 00330035 00320038 00310033 '4.5.5.3.8.2.3.1.' 55 0x00000030: 00000000 '.... ') 56 """ 57 output = ''.join(re.findall(r"'(.*)'", output)) 58 return re.sub(r'[.\s]', '', output) 59 60 61class AdbProxy(object): 62 """Proxy class for ADB. 63 64 For syntactic reasons, the '-' in adb commands need to be replaced with 65 '_'. Can directly execute adb commands on an object: 66 >> adb = AdbProxy(<serial>) 67 >> adb.start_server() 68 >> adb.devices() # will return the console output of "adb devices". 69 """ 70 71 def __init__(self, serial="", ssh_connection=None): 72 """Construct an instance of AdbProxy. 73 74 Args: 75 serial: str serial number of Android device from `adb devices` 76 ssh_connection: SshConnection instance if the Android device is 77 connected to a remote host that we can reach via SSH. 78 """ 79 self.serial = serial 80 self._server_local_port = None 81 adb_path = shutil.which('adb') 82 adb_cmd = [shlex.quote(adb_path)] 83 if serial: 84 adb_cmd.append("-s %s" % serial) 85 if ssh_connection is not None: 86 # Kill all existing adb processes on the remote host (if any) 87 # Note that if there are none, then pkill exits with non-zero status 88 ssh_connection.run("pkill adb", ignore_status=True) 89 # Copy over the adb binary to a temp dir 90 temp_dir = ssh_connection.run("mktemp -d").stdout.strip() 91 ssh_connection.send_file(adb_path, temp_dir) 92 # Start up a new adb server running as root from the copied binary. 93 remote_adb_cmd = "%s/adb %s root" % (temp_dir, "-s %s" % serial 94 if serial else "") 95 ssh_connection.run(remote_adb_cmd) 96 # Proxy a local port to the adb server port 97 local_port = ssh_connection.create_ssh_tunnel(5037) 98 self._server_local_port = local_port 99 100 if self._server_local_port: 101 adb_cmd.append("-P %d" % local_port) 102 self.adb_str = " ".join(adb_cmd) 103 self._ssh_connection = ssh_connection 104 105 def get_user_id(self): 106 """Returns the adb user. Either 2000 (shell) or 0 (root).""" 107 return self.shell('id -u') 108 109 def is_root(self, user_id=None): 110 """Checks if the user is root. 111 112 Args: 113 user_id: if supplied, the id to check against. 114 Returns: 115 True if the user is root. False otherwise. 116 """ 117 if not user_id: 118 user_id = self.get_user_id() 119 return user_id == ROOT_USER_ID 120 121 def ensure_root(self): 122 """Ensures the user is root after making this call. 123 124 Note that this will still fail if the device is a user build, as root 125 is not accessible from a user build. 126 127 Returns: 128 False if the device is a user build. True otherwise. 129 """ 130 self.ensure_user(ROOT_USER_ID) 131 return self.is_root() 132 133 def ensure_user(self, user_id=SHELL_USER_ID): 134 """Ensures the user is set to the given user. 135 136 Args: 137 user_id: The id of the user. 138 """ 139 if self.is_root(user_id): 140 self.root() 141 else: 142 self.unroot() 143 self.wait_for_device() 144 return self.get_user_id() == user_id 145 146 def _exec_cmd(self, cmd, ignore_status=False, timeout=DEFAULT_ADB_TIMEOUT): 147 """Executes adb commands in a new shell. 148 149 This is specific to executing adb commands. 150 151 Args: 152 cmd: A string or list that is the adb command to execute. 153 154 Returns: 155 The stdout of the adb command. 156 157 Raises: 158 AdbError for errors in ADB operations. 159 AdbCommandError for errors from commands executed through ADB. 160 """ 161 if isinstance(cmd, list): 162 cmd = ' '.join(cmd) 163 result = job.run(cmd, ignore_status=True, timeout=timeout) 164 ret, out, err = result.exit_status, result.stdout, result.stderr 165 166 if any(pattern.match(err) for pattern in 167 [ADB_REGEX, DEVICE_OFFLINE_REGEX, DEVICE_NOT_FOUND_REGEX, 168 CANNOT_BIND_LISTENER_REGEX]): 169 raise AdbError(cmd=cmd, stdout=out, stderr=err, ret_code=ret) 170 if "Result: Parcel" in out: 171 return parsing_parcel_output(out) 172 if ignore_status or (ret == 1 and GREP_REGEX.search(cmd)): 173 return out or err 174 if ret != 0: 175 raise AdbCommandError(cmd=cmd, stdout=out, stderr=err, ret_code=ret) 176 return out 177 178 def _exec_adb_cmd(self, name, arg_str, **kwargs): 179 return self._exec_cmd(' '.join((self.adb_str, name, arg_str)), 180 **kwargs) 181 182 def _exec_cmd_nb(self, cmd, **kwargs): 183 """Executes adb commands in a new shell, non blocking. 184 185 Args: 186 cmds: A string that is the adb command to execute. 187 188 """ 189 return job.run_async(cmd, **kwargs) 190 191 def _exec_adb_cmd_nb(self, name, arg_str, **kwargs): 192 return self._exec_cmd_nb(' '.join((self.adb_str, name, arg_str)), 193 **kwargs) 194 195 def tcp_forward(self, host_port, device_port): 196 """Starts tcp forwarding from localhost to this android device. 197 198 Args: 199 host_port: Port number to use on localhost 200 device_port: Port number to use on the android device. 201 202 Returns: 203 Forwarded port on host as int or command output string on error 204 """ 205 if self._ssh_connection: 206 # We have to hop through a remote host first. 207 # 1) Find some free port on the remote host's localhost 208 # 2) Setup forwarding between that remote port and the requested 209 # device port 210 remote_port = self._ssh_connection.find_free_port() 211 host_port = self._ssh_connection.create_ssh_tunnel( 212 remote_port, local_port=host_port) 213 output = self.forward("tcp:%d tcp:%d" % (host_port, device_port), 214 ignore_status=True) 215 # If hinted_port is 0, the output will be the selected port. 216 # Otherwise, there will be no output upon successfully 217 # forwarding the hinted port. 218 if not output: 219 return host_port 220 try: 221 output_int = int(output) 222 except ValueError: 223 return output 224 return output_int 225 226 def remove_tcp_forward(self, host_port): 227 """Stop tcp forwarding a port from localhost to this android device. 228 229 Args: 230 host_port: Port number to use on localhost 231 """ 232 if self._ssh_connection: 233 remote_port = self._ssh_connection.close_ssh_tunnel(host_port) 234 if remote_port is None: 235 logging.warning("Cannot close unknown forwarded tcp port: %d", 236 host_port) 237 return 238 # The actual port we need to disable via adb is on the remote host. 239 host_port = remote_port 240 self.forward("--remove tcp:%d" % host_port) 241 242 def getprop(self, prop_name): 243 """Get a property of the device. 244 245 This is a convenience wrapper for "adb shell getprop xxx". 246 247 Args: 248 prop_name: A string that is the name of the property to get. 249 250 Returns: 251 A string that is the value of the property, or None if the property 252 doesn't exist. 253 """ 254 return self.shell("getprop %s" % prop_name) 255 256 # TODO: This should be abstracted out into an object like the other shell 257 # command. 258 def shell(self, command, ignore_status=False, timeout=DEFAULT_ADB_TIMEOUT): 259 return self._exec_adb_cmd( 260 'shell', 261 shlex.quote(command), 262 ignore_status=ignore_status, 263 timeout=timeout) 264 265 def shell_nb(self, command): 266 return self._exec_adb_cmd_nb('shell', shlex.quote(command)) 267 268 def __getattr__(self, name): 269 def adb_call(*args, **kwargs): 270 usage_metadata_logger.log_usage(self.__module__, name) 271 clean_name = name.replace('_', '-') 272 if clean_name in ['pull', 'push', 'remount'] and 'timeout' not in kwargs: 273 kwargs['timeout'] = DEFAULT_ADB_PULL_TIMEOUT 274 arg_str = ' '.join(str(elem) for elem in args) 275 return self._exec_adb_cmd(clean_name, arg_str, **kwargs) 276 277 return adb_call 278 279 def get_version_number(self): 280 """Returns the version number of ADB as an int (XX in 1.0.XX). 281 282 Raises: 283 AdbError if the version number is not found/parsable. 284 """ 285 version_output = self.version() 286 match = re.search(ADB_VERSION_REGEX, version_output) 287 288 if not match: 289 logging.error('Unable to capture ADB version from adb version ' 290 'output: %s' % version_output) 291 raise AdbError('adb version', version_output, '', '') 292 return int(match.group(1)) 293