cli,remote: Initial userspace TUN server support

Author: strobecat

pymobiledevice3/cli/lockdown.py

@@ -1,5 +1,6 @@
 import asyncio
 import logging
+import platform
 import plistlib
 from pathlib import Path
 from typing import Optional
@@ -164,12 +165,15 @@ def lockdown_wifi_connections(service_provider: LockdownClient, state):
 @lockdown_group.command('start-tunnel', cls=Command)
 @click.option('--script-mode', is_flag=True,
               help='Show only HOST and port number to allow easy parsing from external shell scripts')
-@sudo_required
+@click.option('--userspace-tun-host', help='Specify a host to listen on to enable userspace TUN')
 def cli_start_tunnel(
-        service_provider: LockdownServiceProvider, script_mode: bool) -> None:
+        service_provider: LockdownServiceProvider, script_mode: bool, userspace_tun_host: Optional[str] = None) -> None:
     """ start tunnel """
+    if userspace_tun_host is None or platform.system() == 'Darwin':
+        sudo_required(lambda: None)()
     service = CoreDeviceTunnelProxy(service_provider)
-    asyncio.run(tunnel_task(service, script_mode=script_mode, secrets=None, protocol=TunnelProtocol.TCP), debug=True)
+    asyncio.run(tunnel_task(service, script_mode=script_mode, secrets=None,
+                protocol=TunnelProtocol.TCP, userspace_tun_host=userspace_tun_host), debug=True)
 
 
 @lockdown_group.command('assistive-touch', cls=Command)

pymobiledevice3/cli/remote.py

@@ -1,6 +1,7 @@
 import asyncio
 import dataclasses
 import logging
+import platform
 import sys
 import tempfile
 from functools import partial
@@ -113,30 +114,38 @@ def rsd_info(service_provider: RemoteServiceDiscoveryService):
 
 
 async def tunnel_task(
-        service, secrets: Optional[TextIO] = None, script_mode: bool = False,
+        service, secrets: Optional[TextIO] = None, script_mode: bool = False, userspace_tun_host: Optional[str] = None,
         max_idle_timeout: float = MAX_IDLE_TIMEOUT, protocol: TunnelProtocol = TunnelProtocol.DEFAULT) -> None:
     async with start_tunnel(
-            service, secrets=secrets, max_idle_timeout=max_idle_timeout, protocol=protocol) as tunnel_result:
+            service, secrets=secrets, max_idle_timeout=max_idle_timeout,
+            protocol=protocol, userspace_tun_host=userspace_tun_host) as tunnel_result:
         logger.info('tunnel created')
+        use_userspace_tun = userspace_tun_host is not None
         if script_mode:
-            print(f'{tunnel_result.address} {tunnel_result.port}')
+            if userspace_tun_host:
+                print(f'{tunnel_result.address} {tunnel_result.port} {tunnel_result.interface}')
+            else:
+                print(f'{tunnel_result.address} {tunnel_result.port}')
         else:
+            interface_name = "Userspace TUN" if use_userspace_tun else tunnel_result.interface
+            userspace_param = f"--userspace-tun {tunnel_result.interface} " if use_userspace_tun else ""
             if user_requested_colored_output():
                 if secrets is not None:
                     print(click.style('Secrets: ', bold=True, fg='magenta') +
                           click.style(secrets.name, bold=True, fg='white'))
                 print(click.style('Identifier: ', bold=True, fg='yellow') +
                       click.style(service.remote_identifier, bold=True, fg='white'))
                 print(click.style('Interface: ', bold=True, fg='yellow') +
-                      click.style(tunnel_result.interface, bold=True, fg='white'))
+                      click.style(interface_name, bold=True, fg='white'))
                 print(click.style('Protocol: ', bold=True, fg='yellow') +
                       click.style(tunnel_result.protocol, bold=True, fg='white'))
                 print(click.style('RSD Address: ', bold=True, fg='yellow') +
                       click.style(tunnel_result.address, bold=True, fg='white'))
                 print(click.style('RSD Port: ', bold=True, fg='yellow') +
                       click.style(tunnel_result.port, bold=True, fg='white'))
                 print(click.style('Use the follow connection option:\n', bold=True, fg='yellow') +
-                      click.style(f'--rsd {tunnel_result.address} {tunnel_result.port}', bold=True, fg='cyan'))
+                      click.style(userspace_param + f'--rsd {tunnel_result.address} {tunnel_result.port}',
+                                  bold=True, fg='cyan'))
             else:
                 if secrets is not None:
                     print(f'Secrets: {secrets.name}')
@@ -145,16 +154,17 @@ async def tunnel_task(
                 print(f'Protocol: {tunnel_result.protocol}')
                 print(f'RSD Address: {tunnel_result.address}')
                 print(f'RSD Port: {tunnel_result.port}')
-                print(f'Use the follow connection option:\n'
-                      f'--rsd {tunnel_result.address} {tunnel_result.port}')
+                print('Use the follow connection option:\n' +
+                      userspace_param + f'--rsd {tunnel_result.address} {tunnel_result.port}')
         sys.stdout.flush()
         await tunnel_result.client.wait_closed()
         logger.info('tunnel was closed')
 
 
 async def start_tunnel_task(
         connection_type: ConnectionType, secrets: TextIO, udid: Optional[str] = None, script_mode: bool = False,
-        max_idle_timeout: float = MAX_IDLE_TIMEOUT, protocol: TunnelProtocol = TunnelProtocol.DEFAULT) -> None:
+        max_idle_timeout: float = MAX_IDLE_TIMEOUT, protocol: TunnelProtocol = TunnelProtocol.DEFAULT,
+        userspace_tun_host: Optional[str] = None) -> None:
     if start_tunnel is None:
         raise NotImplementedError('failed to start the tunnel on your platform')
     get_tunnel_services = {
@@ -173,13 +183,14 @@ async def start_tunnel_task(
         service = prompt_device_list(tunnel_services)
 
     await tunnel_task(service, secrets=secrets, script_mode=script_mode, max_idle_timeout=max_idle_timeout,
-                      protocol=protocol)
+                      protocol=protocol, userspace_tun_host=userspace_tun_host)
 
 
 @remote_cli.command('start-tunnel', cls=BaseCommand)
 @click.option('-t', '--connection-type', type=click.Choice([e.value for e in ConnectionType], case_sensitive=False),
               default=ConnectionType.USB.value)
 @click.option('--udid', help='UDID for a specific device to look for')
+@click.option('--userspace-tun-host', help='Specify a host to listen on to enable userspace TUN')
 @click.option('--secrets', type=click.File('wt'), help='TLS keyfile for decrypting with Wireshark')
 @click.option('--script-mode', is_flag=True,
               help='Show only HOST and port number to allow easy parsing from external shell scripts')
@@ -188,17 +199,18 @@ async def start_tunnel_task(
 @click.option('-p', '--protocol',
               type=click.Choice([e.value for e in TunnelProtocol], case_sensitive=False),
               default=TunnelProtocol.DEFAULT.value)
-@sudo_required
 def cli_start_tunnel(
-        connection_type: ConnectionType, udid: Optional[str], secrets: TextIO, script_mode: bool,
-        max_idle_timeout: float, protocol: str) -> None:
+        connection_type: ConnectionType, udid: Optional[str], userspace_tun_host: Optional[str],
+        secrets: TextIO, script_mode: bool, max_idle_timeout: float, protocol: str) -> None:
     """ start tunnel """
     if not verify_tunnel_imports():
         return
+    if userspace_tun_host is None or platform.system() == 'Darwin':
+        sudo_required(lambda: None)()
     asyncio.run(
         start_tunnel_task(
             ConnectionType(connection_type), secrets, udid, script_mode, max_idle_timeout=max_idle_timeout,
-            protocol=TunnelProtocol(protocol)), debug=True)
+            protocol=TunnelProtocol(protocol), userspace_tun_host=userspace_tun_host), debug=True)
 
 
 @dataclasses.dataclass

pymobiledevice3/remote/module_imports.py

@@ -4,9 +4,9 @@
 logger = logging.getLogger(__name__)
 
 try:
-    from pymobiledevice3.remote.tunnel_service import RemotePairingQuicTunnel, start_tunnel
+    from pymobiledevice3.remote.tunnel_service import _RemotePairingQuicTunnelMixin, start_tunnel
 
-    MAX_IDLE_TIMEOUT = RemotePairingQuicTunnel.MAX_IDLE_TIMEOUT
+    MAX_IDLE_TIMEOUT = _RemotePairingQuicTunnelMixin.MAX_IDLE_TIMEOUT
 except ImportError:
     start_tunnel: Optional[Callable] = None
     MAX_IDLE_TIMEOUT = None