diff --git a/pymobiledevice3/__main__.py b/pymobiledevice3/__main__.py index c3aa7a4c1..a2e5f2976 100644 --- a/pymobiledevice3/__main__.py +++ b/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: diff --git a/pymobiledevice3/cli/developer.py b/pymobiledevice3/cli/developer.py index f5da4f3f4..b2daf1b80 100644 --- a/pymobiledevice3/cli/developer.py +++ b/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) diff --git a/pymobiledevice3/cli/remote.py b/pymobiledevice3/cli/remote.py new file mode 100644 index 000000000..f3e37a367 --- /dev/null +++ b/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)