Merge pull request #1103 from doronz88/feature/file-transfer

core-device: implement dirlist and read-file

pymobiledevice3/cli/developer.py

@@ -11,7 +11,7 @@
 from dataclasses import asdict
 from datetime import datetime
 from pathlib import Path
-from typing import List, Optional, Tuple
+from typing import IO, List, Optional, Tuple
 
 import click
 from click.exceptions import MissingParameter, UsageError
@@ -28,6 +28,7 @@
 from pymobiledevice3.osu.os_utils import get_os_utils
 from pymobiledevice3.remote.core_device.app_service import AppServiceService
 from pymobiledevice3.remote.core_device.device_info import DeviceInfoService
+from pymobiledevice3.remote.core_device.file_service import APPLE_DOMAIN_DICT, FileServiceService
 from pymobiledevice3.remote.remote_service_discovery import RemoteServiceDiscoveryService
 from pymobiledevice3.services.accessibilityaudit import AccessibilityAudit
 from pymobiledevice3.services.debugserver_applist import DebugServerAppList
@@ -53,6 +54,7 @@
 from pymobiledevice3.services.screenshot import ScreenshotService
 from pymobiledevice3.services.simulate_location import DtSimulateLocation
 from pymobiledevice3.tcp_forwarder import LockdownTcpForwarder
+from pymobiledevice3.utils import try_decode
 
 OSUTILS = get_os_utils()
 BSC_SUBCLASS = 0x40c
@@ -1048,6 +1050,41 @@ def core_device() -> None:
  pass
 
 
+async def core_device_list_directory_task(
+ service_provider: RemoteServiceDiscoveryService, domain: str, path: str) -> None:
+ async with FileServiceService(service_provider, APPLE_DOMAIN_DICT[domain]) as file_service:
+ print_json(await file_service.retrieve_directory_list(path))
+
+
+@core_device.command('list-directory', cls=RSDCommand)
+@click.argument('domain', type=click.Choice(APPLE_DOMAIN_DICT.keys()))
+@click.argument('path')
+def core_device_list_directory(
+ service_provider: RemoteServiceDiscoveryService, domain: str, path: str) -> None:
+ """ List directory at given domain-path """
+ asyncio.run(core_device_list_directory_task(service_provider, domain, path))
+
+
+async def core_device_read_file_task(
+ service_provider: RemoteServiceDiscoveryService, domain: str, path: str, output: Optional[IO]) -> None:
+ async with FileServiceService(service_provider, APPLE_DOMAIN_DICT[domain]) as file_service:
+ buf = await file_service.retrieve_file(path)
+ if output is not None:
+ output.write(buf)
+ else:
+ print(try_decode(buf))
+
+
+@core_device.command('read-file', cls=RSDCommand)
+@click.argument('domain', type=click.Choice(APPLE_DOMAIN_DICT.keys()))
+@click.argument('path')
+@click.option('-o', '--output', type=click.File('wb'))
+def core_device_read_file(
+ service_provider: RemoteServiceDiscoveryService, domain: str, path: str, output: Optional[IO]) -> None:
+ """ Read file from given domain-path """
+ asyncio.run(core_device_read_file_task(service_provider, domain, path, output))
+
+
 async def core_device_list_launch_application_task(
  service_provider: RemoteServiceDiscoveryService, bundle_identifier: str, argument: List[str],
  kill_existing: bool, suspended: bool, env: List[Tuple[str, str]]) -> None:

pymobiledevice3/remote/core_device/file_service.py

@@ -0,0 +1,71 @@
+import asyncio
+import struct
+import uuid
+from enum import IntEnum
+from typing import AsyncGenerator, List, Mapping, Optional
+
+from pymobiledevice3.exceptions import CoreDeviceError
+from pymobiledevice3.remote.core_device.core_device_service import CoreDeviceService
+from pymobiledevice3.remote.remote_service_discovery import RemoteServiceDiscoveryService
+from pymobiledevice3.remote.xpc_message import XpcUInt64Type
+
+
+class Domain(IntEnum):
+ APP_DATA_CONTAINER = 1
+ APP_GROUP_DATA_CONTAINER = 2
+ TEMPORARY = 3
+ SYSTEM_CRASH_LOGS = 5
+
+
+APPLE_DOMAIN_DICT = {
+ 'appDataContainer': Domain.APP_DATA_CONTAINER,
+ 'appGroupDataContainer': Domain.APP_GROUP_DATA_CONTAINER,
+ 'temporary': Domain.TEMPORARY,
+ 'systemCrashLogs': Domain.SYSTEM_CRASH_LOGS
+}
+
+
+class FileServiceService(CoreDeviceService):
+ """
+ Filesystem control
+ """
+
+ CTRL_SERVICE_NAME = 'com.apple.coredevice.fileservice.control'
+
+ def __init__(self, rsd: RemoteServiceDiscoveryService, domain: Domain) -> None:
+ super().__init__(rsd, self.CTRL_SERVICE_NAME)
+ self.domain: Domain = domain
+ self.session: Optional[str] = None
+
+ async def connect(self) -> None:
+ await super().connect()
+ response = await self.send_receive_request({
+ 'Cmd': 'CreateSession', 'Domain': XpcUInt64Type(self.domain), 'Identifier': '', 'Session': '',
+ 'User': 'mobile'})
+ self.session = response['NewSessionID']
+
+ async def retrieve_directory_list(self, path: str = '.') -> AsyncGenerator[List[str], None]:
+ return (await self.send_receive_request({
+ 'Cmd': 'RetrieveDirectoryList', 'MessageUUID': str(uuid.uuid4()), 'Path': path, 'SessionID': self.session}
+ ))['FileList']
+
+ async def retrieve_file(self, path: str = '.') -> bytes:
+ response = await self.send_receive_request({
+ 'Cmd': 'RetrieveFile', 'Path': path, 'SessionID': self.session}
+ )
+ data_service = self.rsd.get_service_port('com.apple.coredevice.fileservice.data')
+ reader, writer = await asyncio.open_connection(self.service.address[0], data_service)
+ writer.write(b'rwb!FILE' + struct.pack('>QQQQ', response['Response'], 0, response['NewFileID'], 0))
+ await writer.drain()
+ await reader.readexactly(0x24)
+ return await reader.readexactly(struct.unpack('>I', await reader.readexactly(4))[0])
+
+ async def send_receive_request(self, request: Mapping) -> Mapping:
+ response = await self.service.send_receive_request(request)
+ encoded_error = response.get('EncodedError')
+ if encoded_error is not None:
+ localized_description = response.get('LocalizedDescription')
+ if localized_description is not None:
+ raise CoreDeviceError(localized_description)
+ raise CoreDeviceError(encoded_error)
+ return response