From 73206e37116629a5b638e131ff391e72ef30e100 Mon Sep 17 00:00:00 2001 From: Peter Harris Date: Mon, 6 Jan 2025 13:01:22 +0000 Subject: [PATCH] Add remaining Android filesystem helpers --- lglpy/android/adb.py | 61 +++++++++++- lglpy/android/filesystem.py | 104 +++++++++++++++++---- lglpy/android/test.py | 181 ++++++++++++++++++++++++++++++++++-- lglpy/android/utils.py | 8 +- 4 files changed, 324 insertions(+), 30 deletions(-) diff --git a/lglpy/android/adb.py b/lglpy/android/adb.py index f7f2d44..a41fc2d 100644 --- a/lglpy/android/adb.py +++ b/lglpy/android/adb.py @@ -57,6 +57,26 @@ def __init__(self, device: Optional[str] = None, self.device = device self.package = package + def set_device(self, device: str) -> None: + ''' + Set the device for this connection. + + Args: + device: The device identifier, as returned by 'adb devices', or + None for non-specific use. + ''' + self.device = device + + def set_package(self, package: str) -> None: + ''' + Set the package for this connection. + + Args: + package: The package name, as returned by `adb shell pm list + packages` or None for non-specific use. + ''' + self.package = package + def get_base_command(self, args: Iterable[str]) -> list[str]: ''' Get the root of an adb command, injecting device selector if needed. @@ -194,6 +214,44 @@ def adb_async(self, *args: str, text: bool = True, shell: bool = False, # Return the output process a user can use to wait, if needed. return process + def adb_run(self, *args: str, text: bool = True, shell: bool = False, + quote: bool = False, check: bool = True) -> str: + ''' + Call adb to synchronously run a device shell command as the Android + "shell" user, check its result, and capture its output if successful. + + Commands can invoke adb directly, or via the host shell if invoked with + shell=True. When using shell=True on Unix hosts the arguments are + always quoted unless the argument is a '>' redirect shell argument. On + Windows beware of the security implications of the lack of quoting. + + Args: + *args: List of command line parameters. + text: True if output is text, False if binary + shell: True if this should invoke via host shell, False if direct. + quote: True if arguments are quoted, False if unquoted. + check: True if result is checked, False if ignored. + + Returns: + The stdout response written by adb. + + Raises: + CalledProcessError: The invoked call failed. + ''' + # Build the command list + commands = self.get_base_command(['shell']) + commands.extend(args) + packed_commands = self.pack_commands(commands, shell, quote) + + # Invoke the command + rep = sp.run(packed_commands, check=check, shell=shell, text=text, + stdin=sp.DEVNULL, + stdout=sp.PIPE, + stderr=sp.PIPE) + + # Return the output + return rep.stdout + def adb_runas(self, *args: str, text: bool = True, shell: bool = False, quote: bool = False, check: bool = True) -> str: ''' @@ -218,7 +276,8 @@ def adb_runas(self, *args: str, text: bool = True, shell: bool = False, Raises: CalledProcessError: The invoked call failed. ''' - assert self.package, 'Cannot use run-as without a package' + assert self.package, \ + 'Cannot use adb_runas() without package' # Build the command list commands = self.get_base_command(['shell', 'run-as', self.package]) diff --git a/lglpy/android/filesystem.py b/lglpy/android/filesystem.py index 57fd132..9c32a1f 100644 --- a/lglpy/android/filesystem.py +++ b/lglpy/android/filesystem.py @@ -31,6 +31,7 @@ import subprocess as sp from .adb import ADBConnect +from .utils import AndroidUtils class AndroidFilesystem: @@ -108,14 +109,16 @@ def pull_file_from_tmp( conn.adb('pull', device_path, host_dir) if delete: - cls.delete_file_in_tmp(conn, file_name) + cls.delete_file_from_tmp(conn, file_name) except sp.CalledProcessError: return False return True @classmethod - def delete_file_in_tmp(cls, conn: ADBConnect, file_name: str) -> bool: + def delete_file_from_tmp( + cls, conn: ADBConnect, file_name: str, + error_ok: bool = False) -> bool: ''' Delete a file from the device temp directory. @@ -124,6 +127,7 @@ def delete_file_in_tmp(cls, conn: ADBConnect, file_name: str) -> bool: Args: conn: The adb connection. file_name: The name of the file to delete. + error_ok: Ignore errors if the file doesn't exist. Returns: True if the file was deleted, False otherwise. @@ -131,52 +135,118 @@ def delete_file_in_tmp(cls, conn: ADBConnect, file_name: str) -> bool: device_path = posixpath.join(cls.TEMP_DIR, file_name) try: - # Remove old file to prevent false success - conn.adb('shell', 'rm', '-f', device_path) + if error_ok: + conn.adb('shell', 'rm', '-f', device_path) + else: + conn.adb('shell', 'rm', device_path) except sp.CalledProcessError: return False return True - @staticmethod + @classmethod def push_file_to_package( - conn: ADBConnect, package: str, host_path: str, + cls, conn: ADBConnect, host_path: str, executable: bool = False) -> bool: ''' - Push a file to the package data directory. + Push a file to the connection package directory. File will be copied to, e.g.: /data/user/0// Args: conn: The adb connection. - package: The name of the package. host_path: The path of the file on the host file system. executable: True if the file should be configured as executable. Returns: True if the file was copied, False otherwise. ''' - # TODO - return False + assert conn.package, \ + 'Cannot use push_file_to_package() without package' + + # Determine the paths that we need + file_name = os.path.basename(host_path) + tmp_path = posixpath.join(cls.TEMP_DIR, file_name) + + # Copy file to the temp directory + success = cls.push_file_to_tmp(conn, host_path, executable) + if not success: + return False + + # Copy file to the package directory + try: + conn.adb_runas('cp', tmp_path, '.') + except sp.CalledProcessError: + return False + + # Delete the temp file copy + cls.delete_file_from_tmp(conn, file_name) - @staticmethod + return True + + @classmethod def pull_file_from_package( - conn: ADBConnect, package: str, - src_file: str, host_dir: str) -> bool: + cls, conn: ADBConnect, src_file: str, host_dir: str, + delete: bool = False) -> bool: ''' - Pull a file from the package data directory to a host directory. + Pull a file from the connection package directory to a host directory. File will be copied to: /. Args: conn: The adb connection. - package: The name of the package. src_file: The name of the file in the tmp directory. host_path: The destination directory on the host file system. Host directory will be created if it doesn't exist. + delete: Delete the file on the device after copying it. Returns: True if the file was copied, False otherwise. ''' - # TODO - return False + assert conn.package, \ + 'Cannot use pull_file_from_package() without package' + + host_dir = os.path.abspath(host_dir) + os.makedirs(host_dir, exist_ok=True) + + # You cannot adb pull from a package, even if it's debuggable, so + # this is the non-obvious solution ... + host_file = os.path.join(host_dir, src_file) + try: + conn.adb('exec-out', 'run-as', conn.package, + 'cat', src_file, '>', host_file, + text=False, shell=True) + + if delete: + cls.delete_file_from_package(conn, src_file) + except sp.SubprocessError: + return False + + return True + + @classmethod + def delete_file_from_package( + cls, conn: ADBConnect, file_name: str, + error_ok: bool = False) -> bool: + ''' + Delete a file from the package directory. + + File will be deleted from, e.g.: /data/user/0// + + Args: + conn: The adb connection. + file_name: The name of the file to delete. + error_ok: Ignore errors if the file doesn't exist. + + Returns: + True if the file was deleted, False otherwise. + ''' + try: + if error_ok: + conn.adb_runas('rm', '-f', file_name) + else: + conn.adb_runas('rm', file_name) + except sp.CalledProcessError: + return False + + return True \ No newline at end of file diff --git a/lglpy/android/test.py b/lglpy/android/test.py index ebe3fcc..9707105 100644 --- a/lglpy/android/test.py +++ b/lglpy/android/test.py @@ -134,7 +134,7 @@ def test_sync_quote(self): Test direct adb invocation that needs device-side quoting. ''' device = ADBConnect() - result = device.adb('shell', 'echo', 'a | echo', quote=True) + result = device.adb_run('echo', 'a | echo', quote=True) self.assertEqual(result, 'a | echo\n') @unittest.skipIf(os.name == 'nt', 'Not supported on Windows') @@ -143,7 +143,7 @@ def test_sync_shell_quote(self): Test host shell invocation of adb shell that needs host-side quoting. ''' device = ADBConnect() - result = device.adb('shell', 'echo', 'a | echo', shell=True) + result = device.adb_run('echo', 'a | echo', shell=True) self.assertEqual(result, 'a | echo\n') def test_async(self): @@ -337,9 +337,10 @@ def test_util_package_data_dir(self): # Fetch some packages that we can use packages = AndroidUtils.get_packages(conn, True, False) self.assertGreater(len(packages), 0) + conn.set_package(packages[0]) # Test the package - data_dir = AndroidUtils.get_package_data_dir(conn, packages[0]) + data_dir = AndroidUtils.get_package_data_dir(conn) self.assertTrue(data_dir) @@ -372,11 +373,11 @@ def test_util_copy_to_device_tmp(self): self.assertTrue(success) # Validate it pushed OK - data = conn.adb('shell', 'cat', device_file) + data = conn.adb_run('cat', device_file) self.assertEqual(data.strip(), 'test payload') # Cleanup - success = AndroidFilesystem.delete_file_in_tmp(conn, test_file) + success = AndroidFilesystem.delete_file_from_tmp(conn, test_file) self.assertTrue(success) def test_util_copy_to_device_tmp_exec(self): @@ -394,11 +395,11 @@ def test_util_copy_to_device_tmp_exec(self): self.assertTrue(success) # Validate it pushed OK - data = conn.adb('shell', device_file) + data = conn.adb_run(device_file) self.assertEqual(data.strip(), 'test payload exec') # Cleanup - success = AndroidFilesystem.delete_file_in_tmp(conn, test_file) + success = AndroidFilesystem.delete_file_from_tmp(conn, test_file) self.assertTrue(success) def test_util_copy_from_device_keep(self): @@ -420,7 +421,7 @@ def test_util_copy_from_device_keep(self): self.assertTrue(success) # Cleanup - success = AndroidFilesystem.delete_file_in_tmp(conn, test_file) + success = AndroidFilesystem.delete_file_from_tmp(conn, test_file) self.assertTrue(success) def test_util_copy_from_device_delete(self): @@ -450,7 +451,169 @@ def test_util_copy_from_device_delete(self): # Check the file is deleted - this should fail with self.assertRaises(sp.CalledProcessError): - conn.adb('shell', 'ls', device_path) + conn.adb_run('ls', device_path) + + def test_util_copy_to_package(self): + ''' + Test filesystem copy to package data directory. + ''' + conn = ADBConnect() + + # Fetch some packages that we can use + packages = AndroidUtils.get_packages(conn, True, False) + self.assertGreater(len(packages), 0) + conn.set_package(packages[0]) + + test_file = 'test_data.txt' + test_path = get_script_relative_path(test_file) + + # Push the file + success = AndroidFilesystem.push_file_to_package(conn, test_path) + self.assertTrue(success) + + # Validate it pushed OK + data = conn.adb_runas('ls', test_file) + self.assertEqual(data.strip(), test_file) + + # Cleanup tmp - this should fail because the file does not exist + success = AndroidFilesystem.delete_file_from_tmp(conn, test_file) + self.assertFalse(success) + + def test_util_copy_to_package_exec(self): + ''' + Test filesystem copy to package data directory. + ''' + conn = ADBConnect() + + # Fetch some packages that we can use + packages = AndroidUtils.get_packages(conn, True, False) + self.assertGreater(len(packages), 0) + conn.set_package(packages[0]) + + test_file = './test_data.sh' + test_path = get_script_relative_path(test_file) + + # Push the file + success = AndroidFilesystem.push_file_to_package(conn, test_path, True) + self.assertTrue(success) + + # Validate it pushed OK + data = conn.adb_runas(test_file) + self.assertEqual(data.strip(), 'test payload exec') + + # Cleanup the file + success = AndroidFilesystem.delete_file_from_package(conn, test_file) + self.assertTrue(success) + + def test_util_copy_from_package(self): + ''' + Test filesystem copy from package data directory to host. + ''' + conn = ADBConnect() + + # Fetch some packages that we can use + packages = AndroidUtils.get_packages(conn, True, False) + self.assertGreater(len(packages), 0) + conn.set_package(packages[0]) + + test_file = 'test_data.txt' + test_path = get_script_relative_path(test_file) + + # Push the file + success = AndroidFilesystem.push_file_to_package(conn, test_path) + self.assertTrue(success) + + # Validate it pushed OK + data = conn.adb_runas('ls', test_file) + self.assertEqual(data.strip(), test_file) + + # Copy the file + with tempfile.TemporaryDirectory() as host_dir: + host_file = os.path.join(host_dir, test_file) + + AndroidFilesystem.pull_file_from_package( + conn, test_file, host_dir, False) + + # Read the file and validate that it is correct + with open(host_file, 'r', encoding='utf-8') as handle: + data = handle.read() + + self.assertEqual(data.strip(), 'test payload') + + # Cleanup the file + success = AndroidFilesystem.delete_file_from_package(conn, test_file) + self.assertTrue(success) + + def test_util_move_from_package(self): + ''' + Test filesystem move from package data directory to host. + ''' + conn = ADBConnect() + + # Fetch some packages that we can use + packages = AndroidUtils.get_packages(conn, True, False) + self.assertGreater(len(packages), 0) + conn.set_package(packages[0]) + + test_file = 'test_data.txt' + test_path = get_script_relative_path(test_file) + + # Push the file + success = AndroidFilesystem.push_file_to_package(conn, test_path) + self.assertTrue(success) + + # Validate it pushed OK + data = conn.adb_runas('ls', test_file) + self.assertEqual(data.strip(), test_file) + + # Copy the file + with tempfile.TemporaryDirectory() as host_dir: + host_file = os.path.join(host_dir, test_file) + + AndroidFilesystem.pull_file_from_package( + conn, test_file, host_dir, True) + + # Read the file and validate that it is correct + with open(host_file, 'r', encoding='utf-8') as handle: + data = handle.read() + + self.assertEqual(data.strip(), 'test payload') + + # Cleanup the file - this should fail as we deleted the file earlier + success = AndroidFilesystem.delete_file_from_package(conn, test_file) + self.assertFalse(success) + + def test_util_delete_from_package(self): + ''' + Test filesystem delete from package data directory. + ''' + conn = ADBConnect() + + # Fetch some packages that we can use + packages = AndroidUtils.get_packages(conn, True, False) + self.assertGreater(len(packages), 0) + conn.set_package(packages[0]) + + test_file = 'test_data.txt' + test_path = get_script_relative_path(test_file) + device_file = f'/data/local/tmp/{test_file}' + package_dir = AndroidUtils.get_package_data_dir(conn) + + # Push the file + success = AndroidFilesystem.push_file_to_package(conn, test_path) + self.assertTrue(success) + + # Validate it pushed OK + data = conn.adb_runas('ls', test_file) + self.assertEqual(data.strip(), test_file) + + # Cleanup the file + success = AndroidFilesystem.delete_file_from_package(conn, test_file) + self.assertTrue(success) + + # Validate it was deleted - this should fail + with self.assertRaises(sp.CalledProcessError): + data = conn.adb_runas('ls', test_file) def main(): diff --git a/lglpy/android/utils.py b/lglpy/android/utils.py index 705d94a..9eb0b9d 100644 --- a/lglpy/android/utils.py +++ b/lglpy/android/utils.py @@ -230,7 +230,7 @@ def is_package_32bit(conn: ADBConnect, package: str) -> bool: return False @staticmethod - def get_package_data_dir(conn: ADBConnect, package: str): + def get_package_data_dir(conn: ADBConnect): ''' Get the package data directory on the device filesystem. @@ -240,13 +240,15 @@ def get_package_data_dir(conn: ADBConnect, package: str): Args: conn: The adb connection. - package: The name of the package to test. Returns: The package data directory, or None on error. ''' + assert conn.package, \ + 'Cannot use get_package_data_dir() without package' + try: - package = shlex.quote(package) + package = shlex.quote(conn.package) command = f'dumpsys package {package} | grep dataDir' log = conn.adb('shell', command) return log.replace('dataDir=', '').strip()