Merge pull request #1310 from doronz88/feature/propose-file

file_service: add `propose_empty_file()`

pymobiledevice3/cli/developer.py

@@ -7,6 +7,7 @@
 import shlex
 import signal
 import sys
+import time
 from collections import namedtuple
 from dataclasses import asdict
 from datetime import datetime
@@ -841,6 +842,7 @@ def accessibility_settings_set(service_provider: LockdownClient, setting, value)
     service.set_setting(setting, eval(value))
     OSUTILS.wait_return()
 
+
 @accessibility_settings.command('reset', cls=Command)
 def accessibility_settings_reset(service_provider: LockdownClient):
     """
@@ -849,6 +851,7 @@ def accessibility_settings_reset(service_provider: LockdownClient):
     service = AccessibilityAudit(service_provider)
     service.reset_settings()
 
+
 @accessibility.command('shell', cls=Command)
 def accessibility_shell(service_provider: LockdownClient):
     """ start and ipython accessibility shell """
@@ -1085,6 +1088,29 @@ def core_device_read_file(
     asyncio.run(core_device_read_file_task(service_provider, domain, path, output))
 
 
+async def core_device_propose_empty_file_task(
+        service_provider: RemoteServiceDiscoveryService, domain: str, path: str, file_permissions: int, uid: int,
+        gid: int, creation_time: int, last_modification_time: int) -> None:
+    async with FileServiceService(service_provider, APPLE_DOMAIN_DICT[domain]) as file_service:
+        await file_service.propose_empty_file(path, file_permissions, uid, gid, creation_time, last_modification_time)
+
+
+@core_device.command('propose-empty-file', cls=RSDCommand)
+@click.argument('domain', type=click.Choice(APPLE_DOMAIN_DICT.keys()))
+@click.argument('path')
+@click.option('--file-permissions', type=click.INT, default=0o644)
+@click.option('--uid', type=click.INT, default=501)
+@click.option('--gid', type=click.INT, default=501)
+@click.option('--creation-time', type=click.INT, default=time.time())
+@click.option('--last-modification-time', type=click.INT, default=time.time())
+def core_device_propose_empty_file(
+        service_provider: RemoteServiceDiscoveryService, domain: str, path: str, file_permissions: int, uid: int,
+        gid: int, creation_time: int, last_modification_time: int) -> None:
+    """ Write an empty file to given domain-path """
+    asyncio.run(core_device_propose_empty_file_task(service_provider, domain, path, file_permissions, uid, gid, creation_time,
+                                                    last_modification_time))
+
+
 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

@@ -1,5 +1,6 @@
 import asyncio
 import struct
+import time
 import uuid
 from collections.abc import AsyncGenerator
 from enum import IntEnum
@@ -8,7 +9,7 @@
 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
+from pymobiledevice3.remote.xpc_message import XpcInt64Type, XpcUInt64Type
 
 
 class Domain(IntEnum):
@@ -61,6 +62,20 @@ async def retrieve_file(self, path: str = '.') -> bytes:
         await reader.readexactly(0x24)
         return await reader.readexactly(struct.unpack('>I', await reader.readexactly(4))[0])
 
+    async def propose_empty_file(self, path: str = '.', file_permissions: int = 0o644, uid: int = 501, gid: int = 501,
+                                 creation_time: int = time.time(), last_modification_time: int = time.time()) -> None:
+        """ Request to write an empty file at given path. """
+        await self.send_receive_request({
+            'Cmd': 'ProposeEmptyFile',
+            'FileCreationTime': XpcInt64Type(creation_time),
+            'FileLastModificationTime': XpcInt64Type(last_modification_time),
+            'FilePermissions': XpcInt64Type(file_permissions),
+            'FileOwnerUserID': XpcInt64Type(uid),
+            'FileOwnerGroupID': XpcInt64Type(gid),
+            'Path': path,
+            'SessionID': self.session
+        })
+
     async def send_receive_request(self, request: dict) -> dict:
         response = await self.service.send_receive_request(request)
         encoded_error = response.get('EncodedError')