remote: add windows support (doronz88#569)

Author: Tomicyo

README.md

@@ -176,14 +176,19 @@ with RemoteServiceDiscoveryService((host, port)) as rsd:
 
 ## Working with developer tools (iOS >= 17.0)
 
-> **NOTE:** Currently, this is only supported on macOS
+> **NOTE:** Currently, this is only supported on macOS & Windows
 
 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
 first create [trusted tunnel](misc/RemoteXPC.md#trusted-tunnel) as follows:
 
 ```shell
+# -- On macOS
 sudo python3 -m pymobiledevice3 remote start-tunnel
+
+# -- On windows 
+# Use a "run as administrator" shell 
+python3 -m pymobiledevice3 remote start-tunnel
 ```
 
 The root permissions are required since this will create a new TUN/TAP device which is a high privilege operation.
@@ -218,7 +223,12 @@ device is connected.
 To start the Tunneld Server, use the following command (with root privileges):
 
 ```bash
+# -- On macOS
 sudo python3 -m pymobiledevice3 remote tunneld
+
+# -- On windows 
+# Use a "run as administrator" shell 
+python3 -m pymobiledevice3 remote tunneld
 ```
 
 ### Using Tunneld

pymobiledevice3/__main__.py

@@ -1,4 +1,5 @@
 import logging
+import sys
 import traceback
 
 import click
@@ -130,7 +131,10 @@ def main() -> None:
     except PasswordRequiredError:
         logger.error('Device is password protected. Please unlock and retry')
     except AccessDeniedError:
-        logger.error('This command requires root privileges. Consider retrying with "sudo".')
+        if sys.platform == 'win32':
+            logger.error('This command requires admin privileges. Consider retrying with "run-as administrator".')
+        else:
+            logger.error('This command requires root privileges. Consider retrying with "sudo".')
     except BrokenPipeError:
         traceback.print_exc()
     except TunneldConnectionError:

pymobiledevice3/cli/cli_common.py

@@ -105,9 +105,29 @@ def wait_return() -> None:
 UDID_ENV_VAR = 'PYMOBILEDEVICE3_UDID'
 
 
+def is_admin_user() -> bool:
+    """ Check if the current OS user is an Administrator or root.
+
+    See: https://github.com/Preston-Landers/pyuac/blob/master/pyuac/admin.py
+
+    :return: True if the current user is an 'Administrator', otherwise False.
+    """
+    if os.name == 'nt':
+        import win32security
+
+        try:
+            admin_sid = win32security.CreateWellKnownSid(win32security.WinBuiltinAdministratorsSid, None)
+            return win32security.CheckTokenMembership(None, admin_sid)
+        except Exception:
+            return False
+    else:
+        # Check for root on Posix
+        return os.getuid() == 0
+
+
 def sudo_required(func):
     def wrapper(*args, **kwargs):
-        if sys.platform != 'win32' and os.geteuid() != 0:
+        if not is_admin_user():
             raise AccessDeniedError()
         else:
             func(*args, **kwargs)

pymobiledevice3/cli/remote.py

@@ -21,6 +21,12 @@
 logger = logging.getLogger(__name__)
 
 
+def install_driver_if_required() -> None:
+    if sys.platform == 'win32':
+        import pywintunx_pmd3
+        pywintunx_pmd3.install_wetest_driver()
+
+
 def get_device_list() -> List[RemoteServiceDiscoveryService]:
     result = []
     with stop_remoted():
@@ -57,6 +63,7 @@ def cli_tunneld(host: str, port: int, daemonize: bool, protocol: str):
     """ 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)
     if daemonize:
@@ -77,6 +84,7 @@ def cli_tunneld(host: str, port: int, daemonize: bool, protocol: str):
 @click.option('--color/--no-color', default=True)
 def browse(color: bool):
     """ browse devices using bonjour """
+    install_driver_if_required()
     devices = []
     for rsd in get_device_list():
         devices.append({'address': rsd.service.address[0],
@@ -91,6 +99,7 @@ def browse(color: bool):
 @click.option('--color/--no-color', default=True)
 def rsd_info(service_provider: RemoteServiceDiscoveryService, color: bool):
     """ show info extracted from RSD peer """
+    install_driver_if_required()
     print_json(service_provider.peer_info, colored=color)
 
 
@@ -168,6 +177,7 @@ def select_device(udid: str) -> RemoteServiceDiscoveryService:
 @sudo_required
 def cli_start_tunnel(udid: str, secrets: TextIO, script_mode: bool, max_idle_timeout: float, protocol: str):
     """ start quic tunnel """
+    install_driver_if_required()
     protocol = TunnelProtocol(protocol)
     if not verify_tunnel_imports():
         return
@@ -190,5 +200,6 @@ 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()

pymobiledevice3/remote/bonjour.py

@@ -1,4 +1,5 @@
 import dataclasses
+import sys
 import time
 from socket import AF_INET6, inet_ntop
 from typing import List
@@ -7,7 +8,7 @@
 from zeroconf import ServiceBrowser, ServiceListener, Zeroconf
 from zeroconf.const import _TYPE_AAAA
 
-DEFAULT_BONJOUR_TIMEOUT = 1
+DEFAULT_BONJOUR_TIMEOUT = 1 if sys.platform != 'win32' else 2  # On Windows, it takes longer to get the addresses
 
 
 class RemotedListener(ServiceListener):
@@ -46,7 +47,10 @@ def query_bonjour(ip: str) -> BonjourQuery:
 
 
 def get_remoted_addresses(timeout: int = DEFAULT_BONJOUR_TIMEOUT) -> List[str]:
-    ips = [f'{adapter.ips[0].ip[0]}%{adapter.nice_name}' for adapter in get_adapters() if adapter.ips[0].is_IPv6]
+    if sys.platform == 'win32':
+        ips = [f'{adapter.ips[0].ip[0]}%{adapter.ips[0].ip[2]}' for adapter in get_adapters() if adapter.ips[0].is_IPv6]
+    else:
+        ips = [f'{adapter.ips[0].ip[0]}%{adapter.nice_name}' for adapter in get_adapters() if adapter.ips[0].is_IPv6]
     bonjour_queries = [query_bonjour(adapter) for adapter in ips]
     time.sleep(timeout)
     addresses = []

pymobiledevice3/remote/core_device_tunnel_service.py

@@ -41,7 +41,8 @@
 if sys.platform != 'win32':
     from pytun_pmd3 import TunTapDevice
 else:
-    TunTapDevice = None
+    from pywintunx_pmd3 import TunTapDevice, set_logger
+
 from qh3.asyncio import QuicConnectionProtocol
 from qh3.asyncio.client import connect as aioquic_connect
 from qh3.asyncio.protocol import QuicStreamHandler
@@ -73,6 +74,12 @@
 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)
+
 IPV6_HEADER_SIZE = 40
 UDP_HEADER_SIZE = 8
 
@@ -152,12 +159,18 @@ async def wait_closed(self) -> None:
     @asyncio_print_traceback
     async def tun_read_task(self) -> None:
         read_size = self.tun.mtu + len(LOOKBACK_HEADER)
-        async with aiofiles.open(self.tun.fileno(), 'rb', opener=lambda path, flags: path, buffering=0) as f:
+        if sys.platform != 'win32':
+            async with aiofiles.open(self.tun.fileno(), 'rb', opener=lambda path, flags: path, buffering=0) as f:
+                while True:
+                    packet = await f.read(read_size)
+                    assert packet.startswith(LOOKBACK_HEADER)
+                    packet = packet[len(LOOKBACK_HEADER):]
+                    await self.send_packet_to_device(packet)
+        else:
             while True:
-                packet = await f.read(read_size)
-                assert packet.startswith(LOOKBACK_HEADER)
-                packet = packet[len(LOOKBACK_HEADER):]
-                await self.send_packet_to_device(packet)
+                packet = await asyncio.get_running_loop().run_in_executor(None, self.tun.read)
+                if packet:
+                    await self.send_packet_to_device(packet)
 
     def start_tunnel(self, address: str, mtu: int) -> None:
         self.tun = TunTapDevice()
@@ -410,7 +423,7 @@ def save_pair_record(self) -> None:
                 'private_key': self.ed25519_private_key.private_bytes_raw(),
                 'remote_unlock_host_key': self.remote_unlock_host_key
             }))
-        if getenv('SUDO_UID'):
+        if getenv('SUDO_UID') and sys.platform != 'win32':
             chown(self.pair_record_path, int(getenv('SUDO_UID')), int(getenv('SUDO_GID')))
 
     @property

pymobiledevice3/remote/module_imports.py

@@ -1,5 +1,4 @@
 import logging
-import sys
 
 logger = logging.getLogger(__name__)
 
@@ -11,11 +10,7 @@
     start_tunnel = None
     MAX_IDLE_TIMEOUT = None
 
-WIN32_IMPORT_ERROR = """Windows platforms are not yet supported for this command. For more info:
-https://github.com/doronz88/pymobiledevice3/issues/569
-"""
-
-GENERAL_IMPORT_ERROR = """Failed to import `start_tunnel`. Possible reasons are:
+GENERAL_IMPORT_ERROR = """Failed to import `start_tunnel`.
 Please file an issue at:
 https://github.com/doronz88/pymobiledevice3/issues/new?assignees=&labels=&projects=&template=bug_report.md&title=
 
@@ -28,8 +23,5 @@
 def verify_tunnel_imports() -> bool:
     if start_tunnel is not None:
         return True
-    if sys.platform == 'win32':
-        logger.error(WIN32_IMPORT_ERROR)
-        return False
     logger.error(GENERAL_IMPORT_ERROR)
     return False

pymobiledevice3/tunneld.py

@@ -3,6 +3,7 @@
 import logging
 import os
 import signal
+import sys
 import traceback
 from contextlib import asynccontextmanager, suppress
 from typing import Dict, List, Optional, Tuple
@@ -45,8 +46,12 @@ def start(self) -> None:
     async def monitor_adapters(self):
         previous_ips = []
         while True:
-            current_ips = [f'{adapter.ips[0].ip[0]}%{adapter.nice_name}' for adapter in get_adapters() if
-                           adapter.ips[0].is_IPv6]
+            if sys.platform == 'win32':
+                current_ips = [f'{adapter.ips[0].ip[0]}%{adapter.ips[0].ip[2]}' for adapter in get_adapters() if
+                               adapter.ips[0].is_IPv6]
+            else:
+                current_ips = [f'{adapter.ips[0].ip[0]}%{adapter.nice_name}' for adapter in get_adapters() if
+                               adapter.ips[0].is_IPv6]
 
             added = [ip for ip in current_ips if ip not in previous_ips]
             removed = [ip for ip in previous_ips if ip not in current_ips]

requirements.txt

@@ -38,6 +38,7 @@ developer_disk_image>=0.0.2
 opack
 psutil
 pytun-pmd3>=1.0.0 ; platform_system != "Windows"
+pywintunx-pmd3>=1.0.2 ; platform_system == "Windows"
 aiofiles
 prompt_toolkit
 sslpsk-pmd3>=1.0.2