Merge pull request #924 from doronz88/feature/win-pytun

Feature/win pytun

README.md

@@ -145,7 +145,9 @@ Commands:
 ## Working with developer tools (iOS >= 17.0)
 
 > **NOTE:** Currently, this is only officially supported on macOS & Windows (up to iOS 17.3.1), but fully supported on
-> all platforms starting at iOS 17.4 using the new lockdown tunnel.
+> all platforms starting at iOS 17.4 using the new lockdown tunnel. For windows interaction with iOS 17.0-17.3.1, you'll
+> need to install the additional WeTest drivers using: `pymobiledevice3 remote install-wetest-drivers` inside an
+> administrator terminal
 
 Starting at iOS 17.0, Apple introduced the new CoreDevice framework to work with iOS devices. This framework relies on
 the [RemoteXPC](misc/RemoteXPC.md) protocol. In order to communicate with the developer services you'll be required to
@@ -164,7 +166,10 @@ first create [trusted tunnel](misc/RemoteXPC.md#trusted-tunnel) in one of the tw
       # NOTE: on windows, use a privileged shell
 
       # starting at iOS 17.4 you can use the much faster lockdown tunnel
-      sudo python3 -m pymobiledevice3 lockdown start-tunnel 
+      sudo python3 -m pymobiledevice3 lockdown start-tunnel
+      
+      # if you need this connection type to be also available over wifi, you can enable it
+      python3 -m pymobiledevice3 lockdown wifi-connections on
 
       # on older iOS version use the following instead
       # you may pass `-t wifi` to force a WiFi tunnel

pymobiledevice3/cli/remote.py

@@ -17,14 +17,13 @@
 from pymobiledevice3.remote.module_imports import MAX_IDLE_TIMEOUT, start_tunnel, verify_tunnel_imports
 from pymobiledevice3.remote.remote_service_discovery import RSD_PORT, RemoteServiceDiscoveryService
 from pymobiledevice3.remote.tunnel_service import get_core_device_tunnel_services, get_remote_pairing_tunnel_services
-from pymobiledevice3.remote.utils import get_rsds, install_driver_if_required
+from pymobiledevice3.remote.utils import get_rsds
 from pymobiledevice3.tunneld import TUNNELD_DEFAULT_ADDRESS, TunneldRunner
 
 logger = logging.getLogger(__name__)
 
 
 async def browse_rsd(timeout: float = DEFAULT_BONJOUR_TIMEOUT) -> List[Mapping]:
-    install_driver_if_required()
     devices = []
     for rsd in await get_rsds(timeout):
         devices.append({'address': rsd.service.address[0],
@@ -78,7 +77,6 @@ def cli_tunneld(
     """ Start Tunneld service for remote tunneling """
     if not verify_tunnel_imports():
         return
-    install_driver_if_required()
     protocol = TunnelProtocol(protocol)
     tunneld_runner = partial(TunneldRunner.create, host, port, protocol=protocol, usb_monitor=usb, wifi_monitor=wifi,
                              usbmux_monitor=usbmux)
@@ -106,7 +104,6 @@ def browse(timeout: float) -> None:
 @remote_cli.command('rsd-info', cls=RSDCommand)
 def rsd_info(service_provider: RemoteServiceDiscoveryService):
     """ show info extracted from RSD peer """
-    install_driver_if_required()
     print_json(service_provider.peer_info)
 
 
@@ -191,8 +188,6 @@ def cli_start_tunnel(
         connection_type: ConnectionType, udid: Optional[str], secrets: TextIO, script_mode: bool,
         max_idle_timeout: float, protocol: str) -> None:
     """ start tunnel """
-    if connection_type == ConnectionType.USB:
-        install_driver_if_required()
     if not verify_tunnel_imports():
         return
     asyncio.run(
@@ -214,6 +209,22 @@ def cli_delete_pair(udid: str):
 @click.argument('service_name')
 def cli_service(service_provider: RemoteServiceDiscoveryService, service_name: str):
     """ start an ipython shell for interacting with given service """
-    install_driver_if_required()
     with service_provider.start_remote_service(service_name) as service:
         service.shell()
+
+
+@remote_cli.command('install-wetest-drivers', cls=BaseCommand)
+@sudo_required
+def cli_install_wetest_drivers() -> None:
+    """ install WeTests drivers (windows-only) """
+    import pywintunx_pmd3
+    pywintunx_pmd3.install_wetest_driver()
+
+
+@remote_cli.command('uninstall-wetest-drivers', cls=BaseCommand)
+@sudo_required
+def cli_uninstall_wetest_drivers() -> None:
+    """ uninstall WeTests drivers (windows-only) """
+    import pywintunx_pmd3
+    pywintunx_pmd3.uninstall_wetest_driver()
+    pywintunx_pmd3.delete_driver()

pymobiledevice3/remote/tunnel_service.py

@@ -14,15 +14,6 @@
 from asyncio import CancelledError, StreamReader, StreamWriter
 from collections import namedtuple
 from contextlib import asynccontextmanager, suppress
-
-from packaging.version import Version
-
-from pymobiledevice3.lockdown_service_provider import LockdownServiceProvider
-from pymobiledevice3.services.lockdown_service import LockdownService
-
-if sys.platform != 'win32':
-    pass
-
 from pathlib import Path
 from socket import AF_INET6, create_connection
 from ssl import VerifyMode
@@ -41,12 +32,8 @@
 from cryptography.hazmat.primitives.ciphers.aead import ChaCha20Poly1305
 from cryptography.hazmat.primitives.kdf.hkdf import HKDF
 from opack import dumps
-
-if sys.platform != 'win32':
-    from pytun_pmd3 import TunTapDevice
-else:
-    from pywintunx_pmd3 import TunTapDevice, set_logger
-
+from packaging.version import Version
+from pytun_pmd3 import TunTapDevice
 from qh3.asyncio import QuicConnectionProtocol
 from qh3.asyncio.client import connect as aioquic_connect
 from qh3.asyncio.protocol import QuicStreamHandler
@@ -57,6 +44,9 @@
 from srptools import SRPClientSession, SRPContext
 from srptools.constants import PRIME_3072, PRIME_3072_GEN
 
+from pymobiledevice3.lockdown_service_provider import LockdownServiceProvider
+from pymobiledevice3.services.lockdown_service import LockdownService
+
 try:
     from sslpsk_pmd3.sslpsk import SSLPSKContext
 except ImportError:
@@ -80,12 +70,6 @@
 else:
     LOOKBACK_HEADER = b'\x00\x00\x86\xdd'
 
-if sys.platform == 'win32':
-    def wintun_logger(level: int, timestamp: int, message: str) -> None:
-        logging.getLogger('wintun').info(message)
-
-    set_logger(wintun_logger)
-
 logger = logging.getLogger(__name__)
 
 IPV6_HEADER_SIZE = 40
@@ -190,8 +174,8 @@ async def tun_read_task(self) -> None:
 
     def start_tunnel(self, address: str, mtu: int) -> None:
         self.tun = TunTapDevice()
-        self.tun.mtu = mtu
         self.tun.addr = address
+        self.tun.mtu = mtu
         self.tun.up()
         self._tun_read_task = asyncio.create_task(self.tun_read_task(), name=f'tun-read-{address}')
 
@@ -200,8 +184,8 @@ async def stop_tunnel(self) -> None:
         self._tun_read_task.cancel()
         with suppress(CancelledError):
             await self._tun_read_task
-        if sys.platform != 'darwin':
-            self.tun.down()
+        self.tun.down()
+        self.tun.close()
         self.tun = None
 
     @staticmethod

pymobiledevice3/remote/utils.py

@@ -1,6 +1,5 @@
 import contextlib
 import platform
-import sys
 from typing import Generator, List, Optional
 
 import psutil
@@ -87,9 +86,3 @@ def stop_remoted() -> Generator[None, None, None]:
         yield
     finally:
         resume_remoted_if_required()
-
-
-def install_driver_if_required() -> None:
-    if sys.platform == 'win32':
-        import pywintunx_pmd3
-        pywintunx_pmd3.install_wetest_driver()

requirements.txt

@@ -37,7 +37,7 @@ qh3>=0.11.5
 developer_disk_image>=0.0.2
 opack
 psutil
-pytun-pmd3==1.0.0 ; platform_system != "Windows"
+pytun-pmd3>=2.0.5
 pywintunx-pmd3>=1.0.2 ; platform_system == "Windows"
 aiofiles
 prompt_toolkit