[otci] add adb key support for adb interface

This commit adds an option for users to set the adb key when connecting
devices via the adb interface.

tools/cp-caps/README.md

@@ -77,9 +77,10 @@ Device Interfaces:
   REF_ADB_USB=<serial_number>    Connect to the reference device via adb usb
   REF_CLI_SERIAL=<serial_device> Connect to the reference device via cli serial port
   REF_SSH=<device_ip>            Connect to the reference device via ssh
+  ADB_KEY=<adb_key>              Full path to the adb private key
 
 Example:
-  DUT_ADB_USB=1169UC2F2T0M95OR REF_CLI_SERIAL=/dev/ttyACM0 python3 ./tools/cp-caps/rcp_caps_test.py -d
+  ADB_KEY=/usr/local/home/.android/adbkey DUT_ADB_USB=1169UC2F2T0M95OR REF_CLI_SERIAL=/dev/ttyACM0 python3 ./tools/cp-caps/rcp_caps_test.py -d
 ```
 
 ### Test Diag Commands

tools/cp-caps/rcp_caps_test.py

@@ -34,7 +34,7 @@
 import textwrap
 import threading
 
-from typing import List
+from typing import List, Optional
 
 import otci
 from otci import OTCI
@@ -647,11 +647,14 @@ def __get_dut_diag_gpio(self) -> int:
     def __get_dut_diag_raw_power_setting(self) -> str:
         return os.getenv('DUT_DIAG_RAW_POWER_SETTING', '112233')
 
+    def __get_adb_key(self) -> Optional[str]:
+        return os.getenv('ADB_KEY', None)
+
     def __connect_dut(self) -> OTCI:
         if os.getenv('DUT_ADB_TCP'):
-            node = otci.connect_otbr_adb_tcp(os.getenv('DUT_ADB_TCP'))
+            node = otci.connect_otbr_adb_tcp(os.getenv('DUT_ADB_TCP'), adb_key=self.__get_adb_key())
         elif os.getenv('DUT_ADB_USB'):
-            node = otci.connect_otbr_adb_usb(os.getenv('DUT_ADB_USB'))
+            node = otci.connect_otbr_adb_usb(os.getenv('DUT_ADB_USB'), adb_key=self.__get_adb_key())
         elif os.getenv('DUT_CLI_SERIAL'):
             node = otci.connect_cli_serial(os.getenv('DUT_CLI_SERIAL'))
         elif os.getenv('DUT_SSH'):
@@ -667,7 +670,7 @@ def __connect_reference_device(self) -> OTCI:
         elif os.getenv('REF_SSH'):
             node = otci.connect_otbr_ssh(os.getenv('REF_SSH'))
         elif os.getenv('REF_ADB_USB'):
-            node = otci.connect_otbr_adb_usb(os.getenv('REF_ADB_USB'))
+            node = otci.connect_otbr_adb_usb(os.getenv('REF_ADB_USB', adb_key=self.__get_adb_key()))
         else:
             self.__fail("Please set REF_CLI_SERIAL, REF_SSH or REF_ADB_USB to connect to the reference device.")
 
@@ -697,9 +700,10 @@ def parse_arguments():
         '  REF_ADB_USB=<serial_number>    Connect to the reference device via adb usb\r\n'
         '  REF_CLI_SERIAL=<serial_device> Connect to the reference device via cli serial port\r\n'
         '  REF_SSH=<device_ip>            Connect to the reference device via ssh\r\n'
+        '  ADB_KEY=<adb_key>              Full path to the adb private key\r\n'
         '\r\n'
         'Example:\r\n'
-        f'  DUT_ADB_USB=1169UC2F2T0M95OR REF_CLI_SERIAL=/dev/ttyACM0 python3 {sys.argv[0]} -d\r\n')
+        f'  ADB_KEY=/usr/local/home/.android/adbkey DUT_ADB_USB=1169UC2F2T0M95OR REF_CLI_SERIAL=/dev/ttyACM0 python3 {sys.argv[0]} -d\r\n')
 
     parser = argparse.ArgumentParser(formatter_class=argparse.RawDescriptionHelpFormatter,
                                      description=description_msg,

tools/otci/otci/command_handlers.py

@@ -303,10 +303,14 @@ class OtbrAdbCommandRunner(OTCommandHandler):
 
     from adb_shell.adb_device import AdbDevice
 
-    def __init__(self, adb: AdbDevice):
+    def __init__(self, adb: AdbDevice, adb_key: Optional[str] = None):
+        from adb_shell.auth.sign_pythonrsa import PythonRSASigner
+
         self.__adb = adb
         self.__line_read_callback = None
-        self.__adb.connect(rsa_keys=None, auth_timeout_s=0.1)
+        rsa_keys = None if adb_key is None else [PythonRSASigner.FromRSAKeyPath(adb_key)]
+
+        self.__adb.connect(rsa_keys=rsa_keys, auth_timeout_s=0.1)
 
     def execute_command(self, cmd: str, timeout: float) -> List[str]:
         sh_cmd = f'ot-ctl {cmd}'
@@ -342,28 +346,28 @@ def set_line_read_callback(self, callback: Optional[Callable[[str], Any]]):
 
 class OtbrAdbTcpCommandRunner(OtbrAdbCommandRunner):
 
-    def __init__(self, host: str, port: int):
+    def __init__(self, host: str, port: int, adb_key: Optional[str] = None):
         from adb_shell.adb_device import AdbDeviceTcp
 
         self.__host = host
         self.__port = port
 
         adb = AdbDeviceTcp(host, port, default_transport_timeout_s=9.0)
-        super(OtbrAdbTcpCommandRunner, self).__init__(adb)
+        super(OtbrAdbTcpCommandRunner, self).__init__(adb, adb_key)
 
     def __repr__(self):
         return f'{self.__host}:{self.__port}'
 
 
 class OtbrAdbUsbCommandRunner(OtbrAdbCommandRunner):
 
-    def __init__(self, serial: str):
+    def __init__(self, serial: str, adb_key: Optional[str] = None):
         from adb_shell.adb_device import AdbDeviceUsb
 
         self.__serial = serial
 
         adb = AdbDeviceUsb(serial, port_path=None, default_transport_timeout_s=9.0)
-        super(OtbrAdbUsbCommandRunner, self).__init__(adb)
+        super(OtbrAdbUsbCommandRunner, self).__init__(adb, adb_key)
 
     def __repr__(self):
         return f'USB:{self.__serial}'

tools/otci/otci/otci.py

@@ -3175,13 +3175,13 @@ def connect_otbr_ssh(host: str, port: int = 22, username='pi', password='raspber
     return OTCI(cmd_handler)
 
 
-def connect_otbr_adb_tcp(host: str, port: int = 5555):
-    cmd_handler = OtbrAdbTcpCommandRunner(host, port)
+def connect_otbr_adb_tcp(host: str, port: int = 5555, adb_key: Optional[str] = None):
+    cmd_handler = OtbrAdbTcpCommandRunner(host, port, adb_key)
     return OTCI(cmd_handler)
 
 
-def connect_otbr_adb_usb(serial: str):
-    cmd_handler = OtbrAdbUsbCommandRunner(serial)
+def connect_otbr_adb_usb(serial: str, adb_key: Optional[str] = None):
+    cmd_handler = OtbrAdbUsbCommandRunner(serial, adb_key)
     return OTCI(cmd_handler)