diff --git a/tidevice3/api.py b/tidevice3/api.py index 0c3fb7a..88962c1 100644 --- a/tidevice3/api.py +++ b/tidevice3/api.py @@ -77,13 +77,13 @@ def list_devices( DEFAULT_TIMEOUT = 60 -def connect_service_provider(udid: Optional[str], force_usbmux: bool = False, usbmux_address: Optional[str] = None) -> LockdownServiceProvider: +def connect_service_provider(udid: Optional[str], force_usbmux: bool = False, usbmux_address: Optional[str] = None, tunneld_port: Optional[int] = 5555) -> LockdownServiceProvider: """Connect to device and return LockdownServiceProvider""" lockdown = create_using_usbmux(serial=udid, usbmux_address=usbmux_address) if force_usbmux: return lockdown if lockdown.product_version >= "17": - return connect_remote_service_discovery_service(lockdown.udid) + return connect_remote_service_discovery_service(lockdown.udid, "http://localhost:%d" % (tunneld_port)) return lockdown diff --git a/tidevice3/cli/cli_common.py b/tidevice3/cli/cli_common.py index 3f1bcbb..e279e00 100644 --- a/tidevice3/cli/cli_common.py +++ b/tidevice3/cli/cli_common.py @@ -6,6 +6,7 @@ from __future__ import annotations import collections +import logging; from functools import update_wrapper import click @@ -14,6 +15,9 @@ from tidevice3.api import connect_service_provider +logger = logging.getLogger(__name__); + + class OrderedGroup(click.Group): def __init__(self, name=None, commands=None, *args, **attrs): super(OrderedGroup, self).__init__(name, commands, *args, **attrs) @@ -24,14 +28,67 @@ def list_commands(self, ctx): return self.commands +class DeprecatedOption(click.Option): # https://stackoverflow.com/a/50402799/12857692 + def __init__(self, *args, **kwargs): + self.deprecated = kwargs.pop("deprecated", False); + self.preferred = kwargs.pop("preferred", None); + super(DeprecatedOption, self).__init__(*args, **kwargs); + # end __init__() +# end class + + +class CommandWithDeprecatedOptions(click.Command): + def make_parser(self, ctx): # Hook 'make_parser()' and during processing check the name used to invoke the option to see if it is preferred + parser = super().make_parser(ctx); + + # get the parser options + options = set(parser._short_opt.values()); + options |= set(parser._long_opt.values()); + for option in options: + if not isinstance(option.obj, DeprecatedOption): + continue; + # end if + + def make_process(an_option): # construct a closure to the parser option processor + orig_process = an_option.process; + opt_deprecated = getattr(an_option.obj, "deprecated", None); + assert opt_deprecated is not None, "Expected `deprecated` value for `{}`".format(an_option.obj.name); + opt_preferred = getattr(an_option.obj, "preferred", None); + opt_name = getattr(an_option.obj, "name", None); + + def process(value, state): # only called if the option is set + if opt_deprecated: + vv = ["--"+opt_name]; + msg = "The '%s' option is deprecated"; + if opt_preferred is not None: + msg = msg+", use '%s' instead"; + vv.append(opt_preferred); + # end if + logger.warning(msg % tuple(vv)); + # end if + return orig_process(value, state); + # end process() + + return process; + # end make_process() + + option.process = make_process(option); + # end for + return parser; + # end make_parser() +# end class + + @click.group(cls=OrderedGroup, context_settings=dict(help_option_names=["-h", "--help"])) @click.option("-u", "--udid", default=None, help="udid of device") +@click.option("tunneld_port", "--tunneld-port", default=5555, help="tunneld listen port") @click.option("usbmux_address", "--usbmux", help=USBMUX_OPTION_HELP) @click.pass_context -def cli(ctx: click.Context, udid: str, usbmux_address: str): +def cli(ctx: click.Context, udid: str, usbmux_address: str, tunneld_port: int): ctx.ensure_object(dict) ctx.obj['udid'] = udid ctx.obj['usbmux_address'] = usbmux_address + ctx.obj["tunneld_port"] = tunneld_port; def pass_service_provider(func): @@ -50,7 +107,8 @@ def pass_rsd(func): def new_func(ctx, *args, **kwargs): udid = ctx.obj['udid'] usbmux_address = ctx.obj['usbmux_address'] - service_provider = connect_service_provider(udid=udid, usbmux_address=usbmux_address) + tunneld_port = ctx.obj["tunneld_port"]; + service_provider = connect_service_provider(udid=udid, usbmux_address=usbmux_address, tunneld_port=tunneld_port) with service_provider: return ctx.invoke(func, service_provider, *args, **kwargs) return update_wrapper(new_func, func) diff --git a/tidevice3/cli/tunneld.py b/tidevice3/cli/tunneld.py index 02bf211..848cff4 100644 --- a/tidevice3/cli/tunneld.py +++ b/tidevice3/cli/tunneld.py @@ -21,7 +21,7 @@ from pymobiledevice3.exceptions import MuxException from pymobiledevice3.osu.os_utils import OsUtils -from tidevice3.cli.cli_common import cli +from tidevice3.cli.cli_common import cli, CommandWithDeprecatedOptions, DeprecatedOption; from tidevice3.cli.list import list_devices from tidevice3.utils.common import threadsafe_function @@ -161,15 +161,16 @@ def run_forever(self): time.sleep(1) -@cli.command(context_settings={"show_default": True}) +@cli.command(context_settings={"show_default": True}, cls=CommandWithDeprecatedOptions) @click.option( "--pmd3-path", "pmd3_path", help="pymobiledevice3 cli path", default=None, ) -@click.option("--port", "port", help="listen port", default=5555) -def tunneld(pmd3_path: str, port: int): +@click.option("--port", "port", help="listen port", default=5555, cls=DeprecatedOption, deprecated=True, preferred="--tunneld-port") +@click.pass_context +def tunneld(ctx: click.Context, pmd3_path: str, port: int): """start server for iOS >= 17 auto start-tunnel, function like pymobiledevice3 remote tunneld""" if not os_utils.is_admin: logger.error("Please run as root(Mac) or administrator(Windows)") @@ -197,7 +198,8 @@ def shutdown(): target=manager.run_forever, daemon=True, name="device_manager" ).start() try: - uvicorn.run(app, host="0.0.0.0", port=port) + tunneld_port = ctx.obj["tunneld_port"]; + uvicorn.run(app, host="0.0.0.0", port=(tunneld_port if tunneld_port!=5555 else port)) finally: logger.info("Shutting down...") manager.shutdown()