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 17import encodings 18import logging 19import shlex 20import shutil 21 22from mobly.controllers.android_device_lib.adb import AdbProxy 23 24ROOT_USER_ID = '0' 25SHELL_USER_ID = '2000' 26UTF_8 = encodings.utf_8.getregentry().name 27 28 29class BlueberryAdbProxy(AdbProxy): 30 """Proxy class for ADB. 31 32 For syntactic reasons, the '-' in adb commands need to be replaced with 33 '_'. Can directly execute adb commands on an object: 34 >> adb = BlueberryAdbProxy(<serial>) 35 >> adb.start_server() 36 >> adb.devices() # will return the console output of "adb devices". 37 """ 38 39 def __init__(self, serial="", ssh_connection=None): 40 """Construct an instance of AdbProxy. 41 42 Args: 43 serial: str serial number of Android device from `adb devices` 44 ssh_connection: SshConnection instance if the Android device is 45 connected to a remote host that we can reach via SSH. 46 """ 47 super().__init__(serial) 48 self._server_local_port = None 49 adb_path = shutil.which('adb') 50 adb_cmd = [shlex.quote(adb_path)] 51 if serial: 52 adb_cmd.append("-s %s" % serial) 53 if ssh_connection is not None: 54 # Kill all existing adb processes on the remote host (if any) 55 # Note that if there are none, then pkill exits with non-zero status 56 ssh_connection.run("pkill adb", ignore_status=True) 57 # Copy over the adb binary to a temp dir 58 temp_dir = ssh_connection.run("mktemp -d").stdout.strip() 59 ssh_connection.send_file(adb_path, temp_dir) 60 # Start up a new adb server running as root from the copied binary. 61 remote_adb_cmd = "%s/adb %s root" % (temp_dir, "-s %s" % serial if serial else "") 62 ssh_connection.run(remote_adb_cmd) 63 # Proxy a local port to the adb server port 64 local_port = ssh_connection.create_ssh_tunnel(5037) 65 self._server_local_port = local_port 66 67 if self._server_local_port: 68 adb_cmd.append("-P %d" % local_port) 69 self.adb_str = " ".join(adb_cmd) 70 self._ssh_connection = ssh_connection 71 72 def get_user_id(self): 73 """Returns the adb user. Either 2000 (shell) or 0 (root).""" 74 return self.shell('id -u').decode(UTF_8).rstrip() 75 76 def is_root(self, user_id=None): 77 """Checks if the user is root. 78 79 Args: 80 user_id: if supplied, the id to check against. 81 Returns: 82 True if the user is root. False otherwise. 83 """ 84 if not user_id: 85 user_id = self.get_user_id() 86 return user_id == ROOT_USER_ID 87 88 def ensure_root(self): 89 """Ensures the user is root after making this call. 90 91 Note that this will still fail if the device is a user build, as root 92 is not accessible from a user build. 93 94 Returns: 95 False if the device is a user build. True otherwise. 96 """ 97 self.ensure_user(ROOT_USER_ID) 98 return self.is_root() 99 100 def ensure_user(self, user_id=SHELL_USER_ID): 101 """Ensures the user is set to the given user. 102 103 Args: 104 user_id: The id of the user. 105 """ 106 if self.is_root(user_id): 107 self.root() 108 else: 109 self.unroot() 110 self.wait_for_device() 111 return self.get_user_id() == user_id 112 113 def tcp_forward(self, host_port, device_port): 114 """Starts tcp forwarding from localhost to this android device. 115 116 Args: 117 host_port: Port number to use on localhost 118 device_port: Port number to use on the android device. 119 120 Returns: 121 Forwarded port on host as int or command output string on error 122 """ 123 if self._ssh_connection: 124 # We have to hop through a remote host first. 125 # 1) Find some free port on the remote host's localhost 126 # 2) Setup forwarding between that remote port and the requested 127 # device port 128 remote_port = self._ssh_connection.find_free_port() 129 host_port = self._ssh_connection.create_ssh_tunnel(remote_port, local_port=host_port) 130 try: 131 output = self.forward(["tcp:%d" % host_port, "tcp:%d" % device_port]) 132 except AdbError as error: 133 return error 134 # If hinted_port is 0, the output will be the selected port. 135 # Otherwise, there will be no output upon successfully 136 # forwarding the hinted port. 137 if not output: 138 return host_port 139 try: 140 output_int = int(output) 141 except ValueError: 142 return output 143 return output_int 144 145 def remove_tcp_forward(self, host_port): 146 """Stop tcp forwarding a port from localhost to this android device. 147 148 Args: 149 host_port: Port number to use on localhost 150 """ 151 if self._ssh_connection: 152 remote_port = self._ssh_connection.close_ssh_tunnel(host_port) 153 if remote_port is None: 154 logging.warning("Cannot close unknown forwarded tcp port: %d", host_port) 155 return 156 # The actual port we need to disable via adb is on the remote host. 157 host_port = remote_port 158 self.forward(["--remove", "tcp:%d" % host_port]) 159 160 def path_exists(self, path): 161 """Check if a file path exists on an Android device 162 163 :param path: file path, could be a directory 164 :return: True if file path exists 165 """ 166 try: 167 ret = self.shell("ls {}".format(path)) 168 if ret is not None and len(ret) > 0: 169 return True 170 else: 171 return False 172 except AdbError as e: 173 logging.debug("path {} does not exist, error={}".format(path, e)) 174 return False 175