Feature/ipsw download

.github/workflows/python-app.yml

@@ -32,17 +32,15 @@ jobs:
         uses: actions/setup-python@v4
         with:
           python-version: ${{ matrix.python-version }}
-      - name: Lint with flake8
+      - name: Install dependencies
         run: |
-          python -m pip install flake8
-          flake8 . --max-line-length=127
-      - name: Verify sorted imports
+          python3 -m pip install --upgrade pip
+          python3 -m pip install pre-commit
+      - name: Run pre-commit hooks
         run: |
-          python -m pip install isort
-          isort . -m HANGING_INDENT -l 120 --check-only
+          pre-commit run --all-files
       - name: Test install
         run: |
-          python -m pip install --upgrade pip
           python -m pip install -U '.[test]'
       - name: Test show usage
         run: pytest -m cli

.pre-commit-config.yaml

@@ -0,0 +1,15 @@
+# .pre-commit-config.yaml
+repos:
+  - repo: https://github.com/pre-commit/mirrors-isort
+    rev: v5.9.3
+    hooks:
+      - id: isort
+        args: [ '-m', 'HANGING_INDENT', '-l', '120','--check-only' ]
+        files: \.py$
+
+  - repo: https://github.com/pycqa/flake8
+    rev: "7.0.0"
+    hooks:
+      - id: flake8
+        args: [ '--max-line-length=127' ]
+        files: \.py$

README.md

@@ -293,8 +293,8 @@ pymobiledevice3 backup2 backup --full DIRECTORY
 # Restore a given backup
 pymobiledevice3 backup2 restore DIRECTORY
 
-# Perform a software upate by a given IPSW file:
-pymobiledevice3 restore update /path/to/ipsw
+# Perform a software upate by a given IPSW file/url:
+pymobiledevice3 restore update -i /path/to/ipsw | url
 
 # Note: The following webinspector subcommands will require the Web Inspector feature to be turned on
 

pymobiledevice3/__main__.py

@@ -8,7 +8,7 @@
 from pymobiledevice3.exceptions import AccessDeniedError, ConnectionFailedToUsbmuxdError, DeprecationError, \
     DeveloperModeError, DeveloperModeIsNotEnabledError, DeviceHasPasscodeSetError, DeviceNotFoundError, \
     FeatureNotSupportedError, InternalError, InvalidServiceError, MessageNotSupportedError, MissingValueError, \
-    NoDeviceConnectedError, NoDeviceSelectedError, NotEnoughDiskSpaceError, NotPairedError, OSNotSupportedError, \
+    NoDeviceConnectedError, NotEnoughDiskSpaceError, NotPairedError, OSNotSupportedError, \
     PairingDialogResponsePendingError, PasswordRequiredError, RSDRequiredError, SetProhibitedError, \
     TunneldConnectionError, UserDeniedPairingError
 from pymobiledevice3.osu.os_utils import get_os_utils
@@ -144,8 +144,6 @@ def main() -> None:
                 sys.argv += ['--tunnel', e.identifier]
             return main()
         logger.error(INVALID_SERVICE_MESSAGE)
-    except NoDeviceSelectedError:
-        return
     except PasswordRequiredError:
         logger.error('Device is password protected. Please unlock and retry')
     except AccessDeniedError:

pymobiledevice3/cli/cli_common.py

@@ -6,7 +6,7 @@
 import sys
 import uuid
 from functools import wraps
-from typing import Callable, List, Mapping, Optional, Tuple
+from typing import Any, Callable, List, Mapping, Optional, Tuple
 
 import click
 import coloredlogs
@@ -16,8 +16,7 @@
 from inquirer3.themes import GreenPassion
 from pygments import formatters, highlight, lexers
 
-from pymobiledevice3.exceptions import AccessDeniedError, DeviceNotFoundError, NoDeviceConnectedError, \
-    NoDeviceSelectedError
+from pymobiledevice3.exceptions import AccessDeniedError, DeviceNotFoundError, NoDeviceConnectedError
 from pymobiledevice3.lockdown import LockdownClient, create_using_usbmux
 from pymobiledevice3.osu.os_utils import get_os_utils
 from pymobiledevice3.remote.remote_service_discovery import RemoteServiceDiscoveryService
@@ -123,13 +122,17 @@ def wrapper(*args, **kwargs):
     return wrapper
 
 
-def prompt_device_list(device_list: List):
-    device_question = [inquirer3.List('device', message='choose device', choices=device_list, carousel=True)]
+def prompt_selection(choices: List[Any], message: str, idx: bool = False) -> Any:
+    question = [inquirer3.List('selection', message=message, choices=choices, carousel=True)]
     try:
-        result = inquirer3.prompt(device_question, theme=GreenPassion(), raise_keyboard_interrupt=True)
-        return result['device']
+        result = inquirer3.prompt(question, theme=GreenPassion(), raise_keyboard_interrupt=True)
     except KeyboardInterrupt:
-        raise NoDeviceSelectedError()
+        raise click.ClickException('No selection was made')
+    return result['selection'] if not idx else choices.index(result['selection'])
+
+
+def prompt_device_list(device_list: List):
+    return prompt_selection(device_list, 'Choose device')
 
 
 def choose_service_provider(callback: Callable):

pymobiledevice3/cli/developer.py

@@ -707,9 +707,6 @@ def fetch_symbols_list(service_provider: LockdownServiceProvider) -> None:
 
 
 async def fetch_symbols_download_task(service_provider: LockdownServiceProvider, out: str) -> None:
-    if not isinstance(service_provider, RemoteServiceDiscoveryService):
-        raise ArgumentError('service_provider must be a RemoteServiceDiscoveryService for iOS 17+ devices')
-
     out = Path(out)
     out.mkdir(parents=True, exist_ok=True)
 
@@ -736,6 +733,8 @@ async def fetch_symbols_download_task(service_provider: LockdownServiceProvider,
                 logger.info(f'writing to: {file}')
                 fetch_symbols.get_file(i, f)
     else:
+        if not isinstance(service_provider, RemoteServiceDiscoveryService):
+            raise ArgumentError('service_provider must be a RemoteServiceDiscoveryService for iOS 17+ devices')
         async with RemoteFetchSymbolsService(service_provider) as fetch_symbols:
             await fetch_symbols.download(out)
 

pymobiledevice3/cli/restore.py

@@ -1,25 +1,29 @@
 import asyncio
+import contextlib
 import logging
 import os
 import plistlib
+import tempfile
 import traceback
-from typing import IO
+from pathlib import Path
+from typing import IO, Generator, Optional, Union
 from zipfile import ZipFile
 
 import click
 import IPython
+import requests
 from pygments import formatters, highlight, lexers
-from remotezip import RemoteZip
 
 from pymobiledevice3 import usbmux
-from pymobiledevice3.cli.cli_common import print_json, set_verbosity
+from pymobiledevice3.cli.cli_common import print_json, prompt_selection, set_verbosity
 from pymobiledevice3.exceptions import ConnectionFailedError, ConnectionFailedToUsbmuxdError, IncorrectModeError
 from pymobiledevice3.irecv import IRecv
 from pymobiledevice3.lockdown import LockdownClient, create_using_usbmux
 from pymobiledevice3.restore.device import Device
 from pymobiledevice3.restore.recovery import Behavior, Recovery
 from pymobiledevice3.restore.restore import Restore
 from pymobiledevice3.services.diagnostics import DiagnosticsService
+from pymobiledevice3.utils import file_download
 
 SHELL_USAGE = """
 # use `irecv` variable to access Restore mode API
@@ -28,6 +32,7 @@
 """
 
 logger = logging.getLogger(__name__)
+IPSWME_API = 'https://api.ipsw.me/v4/device/'
 
 
 class Command(click.Command):
@@ -39,15 +44,18 @@ def __init__(self, *args, **kwargs):
         ]
 
     @staticmethod
-    def device(ctx, param, value):
+    def device(ctx, param, value) -> Optional[Union[LockdownClient, IRecv]]:
         if '_PYMOBILEDEVICE3_COMPLETE' in os.environ:
             # prevent lockdown connection establishment when in autocomplete mode
             return
 
         ecid = value
         logger.debug('searching among connected devices via lockdownd')
+        devices = [dev for dev in usbmux.list_devices() if dev.connection_type == 'USB']
+        if len(devices) > 1:
+            raise click.ClickException('Multiple device detected')
         try:
-            for device in usbmux.list_devices():
+            for device in devices:
                 try:
                     lockdown = create_using_usbmux(serial=device.serial, connection_type='USB')
                 except (ConnectionFailedError, IncorrectModeError):
@@ -64,6 +72,71 @@ def device(ctx, param, value):
         return IRecv(ecid=ecid)
 
 
+@contextlib.contextmanager
+def tempzip_download_ctx(url: str) -> Generator[ZipFile, None, None]:
+    with tempfile.TemporaryDirectory() as tmpdir:
+        tmpzip = Path(tmpdir) / url.split('/')[-1]
+        file_download(url, tmpzip)
+        yield ZipFile(tmpzip)
+
+
+@contextlib.contextmanager
+def zipfile_ctx(path: str) -> Generator[ZipFile, None, None]:
+    yield ZipFile(path)
+
+
+class IPSWCommand(Command):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.params.extend([click.Option(('ipsw_ctx', '-i'), required=False, callback=self.ipsw_ctx),
+                            click.Option(('tss', '--tss'), type=click.File('rb'), callback=self.tss)])
+
+    @staticmethod
+    def ipsw_ctx(ctx, param, value) -> Generator[ZipFile, None, None]:
+        if value and not value.startswith(('http://', 'https://')):
+            return zipfile_ctx(value)
+
+        url = value
+        if url is None:
+            url = query_ipswme(ctx.params['device'].product_type)
+        return tempzip_download_ctx(url)
+
+    @staticmethod
+    def tss(ctx, param, value) -> Optional[IO]:
+        if value is None:
+            return
+        return plistlib.load(value)
+
+
+def query_ipswme(identifier: str) -> str:
+    resp = requests.get(IPSWME_API + identifier, headers={'Accept': 'application/json'})
+    firmwares = resp.json()['firmwares']
+    display_list = [f'{entry["version"]}: {entry["buildid"]}' for entry in firmwares if entry['signed']]
+    idx = prompt_selection(display_list, 'Choose version', idx=True)
+    return firmwares[idx]['url']
+
+
+async def restore_update_task(device: Device, ipsw: ZipFile, tss: Optional[IO], erase: bool, ignore_fdr: bool) -> None:
+    lockdown = None
+    irecv = None
+    if isinstance(device, LockdownClient):
+        lockdown = device
+    elif isinstance(device, IRecv):
+        irecv = device
+    device = Device(lockdown=lockdown, irecv=irecv)
+
+    behavior = Behavior.Update
+    if erase:
+        behavior = Behavior.Erase
+
+    try:
+        await Restore(ipsw, device, tss=tss, behavior=behavior, ignore_fdr=ignore_fdr).update()
+    except Exception:
+        # click may "swallow" several exception types so we try to catch them all here
+        traceback.print_exc()
+        raise
+
+
 @click.group()
 def cli():
     """ cli """
@@ -111,10 +184,9 @@ def restore_restart(device):
         device.reboot()
 
 
-@restore.command('tss', cls=Command)
-@click.argument('ipsw')
+@restore.command('tss', cls=IPSWCommand)
 @click.argument('out', type=click.File('wb'), required=False)
-def restore_tss(device, ipsw, out):
+def restore_tss(device: Device, ipsw_ctx: Generator, out):
     """ query SHSH blobs """
     lockdown = None
     irecv = None
@@ -123,84 +195,41 @@ def restore_tss(device, ipsw, out):
     elif isinstance(device, IRecv):
         irecv = device
 
-    if ipsw.startswith('http://') or ipsw.startswith('https://'):
-        ipsw = RemoteZip(ipsw)
-    else:
-        ipsw = ZipFile(ipsw)
-
     device = Device(lockdown=lockdown, irecv=irecv)
-    tss = Recovery(ipsw, device).fetch_tss_record()
+    with ipsw_ctx as ipsw:
+        tss = Recovery(ipsw, device).fetch_tss_record()
     if out:
         plistlib.dump(tss, out)
     print_json(tss)
 
 
-@restore.command('ramdisk', cls=Command)
-@click.argument('ipsw')
-@click.option('--tss', type=click.File('rb'))
-def restore_ramdisk(device, ipsw, tss):
+@restore.command('ramdisk', cls=IPSWCommand)
+def restore_ramdisk(device: Device, ipsw_ctx: Generator, tss: IO):
     """
     don't perform an actual restore. just enter the update ramdisk
 
     ipsw can be either a filename or an url
     """
-    if tss:
-        tss = plistlib.load(tss)
-
-    if ipsw.startswith('http://') or ipsw.startswith('https://'):
-        ipsw = RemoteZip(ipsw)
-    else:
-        ipsw = ZipFile(ipsw)
-
-    lockdown = None
-    irecv = None
-    if isinstance(device, LockdownClient):
-        lockdown = device
-    elif isinstance(device, IRecv):
-        irecv = device
-    device = Device(lockdown=lockdown, irecv=irecv)
-    Recovery(ipsw, device, tss=tss).boot_ramdisk()
-
-
-async def restore_update_task(device: Device, ipsw: str, tss: IO, erase: bool, ignore_fdr: bool) -> None:
-    if tss:
-        tss = plistlib.load(tss)
-
-    if ipsw.startswith('http://') or ipsw.startswith('https://'):
-        ipsw = RemoteZip(ipsw)
-    else:
-        ipsw = ZipFile(ipsw)
-
     lockdown = None
     irecv = None
     if isinstance(device, LockdownClient):
         lockdown = device
     elif isinstance(device, IRecv):
         irecv = device
     device = Device(lockdown=lockdown, irecv=irecv)
-
-    behavior = Behavior.Update
-    if erase:
-        behavior = Behavior.Erase
-
-    try:
-        await Restore(ipsw, device, tss=tss, behavior=behavior, ignore_fdr=ignore_fdr).update()
-    except Exception:
-        # click may "swallow" several exception types so we try to catch them all here
-        traceback.print_exc()
-        raise
+    with ipsw_ctx as ipsw:
+        Recovery(ipsw, device, tss=tss).boot_ramdisk()
 
 
-@restore.command('update', cls=Command)
-@click.argument('ipsw')
-@click.option('--tss', type=click.File('rb'))
+@restore.command('update', cls=IPSWCommand)
 @click.option('--erase', is_flag=True, help='use the Erase BuildIdentity (full factory-reset)')
 @click.option('--ignore-fdr', is_flag=True, help='only establish an FDR service connection, but don\'t proxy any '
                                                  'traffic')
-def restore_update(device: Device, ipsw: str, tss: IO, erase: bool, ignore_fdr: bool) -> None:
+def restore_update(device: Device, ipsw_ctx: Generator, tss: IO, erase: bool, ignore_fdr: bool) -> None:
     """
     perform an update
 
     ipsw can be either a filename or an url
     """
-    asyncio.run(restore_update_task(device, ipsw, tss, erase, ignore_fdr), debug=True)
+    with ipsw_ctx as ipsw:
+        asyncio.run(restore_update_task(device, ipsw, tss, erase, ignore_fdr), debug=True)

pymobiledevice3/exceptions.py

@@ -10,7 +10,7 @@
     'PairingDialogResponsePendingError', 'UserDeniedPairingError', 'InvalidHostIDError', 'SetProhibitedError',
     'MissingValueError', 'PasscodeRequiredError', 'AmfiError', 'DeviceHasPasscodeSetError', 'NotificationTimeoutError',
     'DeveloperModeError', 'ProfileError', 'IRecvError', 'IRecvNoDeviceConnectedError', 'UnrecognizedSelectorError',
-    'NoDeviceSelectedError', 'MessageNotSupportedError', 'InvalidServiceError', 'InspectorEvaluateError',
+    'MessageNotSupportedError', 'InvalidServiceError', 'InspectorEvaluateError',
     'LaunchingApplicationError', 'BadCommandError', 'BadDevError', 'ConnectionFailedError', 'CoreDeviceError',
     'AccessDeniedError', 'RSDRequiredError', 'SysdiagnoseTimeoutError', 'GetProhibitedError',
     'FeatureNotSupportedError', 'OSNotSupportedError', 'DeprecationError', 'NotEnoughDiskSpaceError'
@@ -290,10 +290,6 @@ class IRecvNoDeviceConnectedError(IRecvError):
     pass
 
 
-class NoDeviceSelectedError(PyMobileDevice3Exception):
-    pass
-
-
 class MessageNotSupportedError(PyMobileDevice3Exception):
     pass