cli: add remote subcommand

pymobiledevice3/__main__.py

@@ -23,6 +23,7 @@
 from pymobiledevice3.cli.processes import cli as ps_cli
 from pymobiledevice3.cli.profile import cli as profile_cli
 from pymobiledevice3.cli.provision import cli as provision_cli
+from pymobiledevice3.cli.remote import cli as remote_cli
 from pymobiledevice3.cli.restore import cli as restore_cli
 from pymobiledevice3.cli.springboard import cli as springboard_cli
 from pymobiledevice3.cli.syslog import cli as syslog_cli
@@ -35,7 +36,9 @@
 
 coloredlogs.install(level=logging.INFO)
 
+logging.getLogger('quic').disabled = True
 logging.getLogger('asyncio').disabled = True
+logging.getLogger('zeroconf').disabled = True
 logging.getLogger('parso.cache').disabled = True
 logging.getLogger('parso.cache.pickle').disabled = True
 logging.getLogger('parso.python.diff').disabled = True
@@ -50,7 +53,8 @@ def cli():
     cli_commands = click.CommandCollection(sources=[
         developer_cli, mounter_cli, apps_cli, profile_cli, lockdown_cli, diagnostics_cli, syslog_cli, pcap_cli,
         crash_cli, afc_cli, ps_cli, notification_cli, usbmux_cli, power_assertion_cli, springboard_cli,
-        provision_cli, backup_cli, restore_cli, activation_cli, companion_cli, webinspector_cli, amfi_cli, bonjour_cli
+        provision_cli, backup_cli, restore_cli, activation_cli, companion_cli, webinspector_cli, amfi_cli, bonjour_cli,
+        remote_cli
     ])
     cli_commands.context_settings = dict(help_option_names=['-h', '--help'])
     try:

pymobiledevice3/cli/developer.py

@@ -222,8 +222,8 @@ def netstat(lockdown: LockdownClient):
             for event in monitor:
                 if isinstance(event, ConnectionDetectionEvent):
                     logger.info(
-                        f'Connection detected: {event.local_address.data.address}:{event.local_address.port} -> '
-                        f'{event.remote_address.data.address}:{event.remote_address.port}')
+                        f'Connection detected: {event.local_address.data.hostname}:{event.local_address.port} -> '
+                        f'{event.remote_address.data.hostname}:{event.remote_address.port}')
 
 
 @dvt.command('screenshot', cls=Command)

pymobiledevice3/cli/remote.py

@@ -0,0 +1,82 @@
+import asyncio
+import logging
+import os
+from typing import Optional
+
+import click
+from cryptography.hazmat.primitives.asymmetric import rsa
+
+from pymobiledevice3.cli.cli_common import UDID_ENV_VAR, print_json, prompt_device_list, set_verbosity
+from pymobiledevice3.remote.core_device_tunnel_service import create_core_device_tunnel_service
+from pymobiledevice3.remote.remote_service_discovery import RSD_PORT, RemoteServiceDiscoveryService, \
+    get_remoted_device, get_remoted_devices
+
+logger = logging.getLogger(__name__)
+
+
+class RemoteCommand(click.Command):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.params[:0] = [
+            click.Option(('hostname', '--udid'), envvar=UDID_ENV_VAR, callback=self.udid,
+                         help=f'Device unique identifier. You may pass {UDID_ENV_VAR} environment variable to pass this'
+                              f' option as well'),
+            click.Option(('verbosity', '-v', '--verbose'), count=True, callback=set_verbosity, expose_value=False),
+        ]
+
+    @staticmethod
+    def udid(ctx, param: str, value: str) -> Optional[str]:
+        if '_PYMOBILEDEVICE3_COMPLETE' in os.environ:
+            # prevent lockdown connection establishment when in autocomplete mode
+            return
+
+        if value is not None:
+            return get_remoted_device(udid=value).hostname
+
+        device_options = get_remoted_devices()
+        if len(device_options) == 1:
+            return device_options[0].hostname
+
+        return prompt_device_list(device_options).hostname
+
+
+@click.group()
+def cli():
+    """ remote cli """
+    pass
+
+
+@cli.group('remote')
+def remote_cli():
+    """ remote options """
+    pass
+
+
+@remote_cli.command('rsd-info', cls=RemoteCommand)
+@click.option('--color/--no-color', default=True)
+def rsd_info(hostname: str, color: bool):
+    """ show info extracted from RSD peer """
+    with RemoteServiceDiscoveryService((hostname, RSD_PORT)) as rsd:
+        print_json(rsd.peer_info, colored=color)
+
+
+@remote_cli.command('create-listener', cls=RemoteCommand)
+@click.option('-p', '--protocol', type=click.Choice(['quic', 'udp']))
+@click.option('--color/--no-color', default=True)
+def create_listener(hostname: str, protocol: str, color: bool):
+    """ start a remote listener """
+    private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+    with RemoteServiceDiscoveryService((hostname, RSD_PORT)) as rsd:
+        with create_core_device_tunnel_service(rsd, autopair=True) as service:
+            print_json(service.create_listener(private_key, protocol=protocol), colored=color)
+
+
+@remote_cli.command('start-quic-tunnel', cls=RemoteCommand)
+@click.option('--color/--no-color', default=True)
+def start_quic_tunnel(hostname: str, color: bool):
+    """ start quic tunnel """
+    logger.critical('This is a WIP command. Will only print the required parameters for the quic connection')
+    private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+    with RemoteServiceDiscoveryService((hostname, RSD_PORT)) as rsd:
+        with create_core_device_tunnel_service(rsd, autopair=True) as service:
+            print_json(asyncio.run(service.start_quic_tunnel(private_key)), colored=color)