fix: add an option for connecting to different tunnel server port

tidevice3/api.py

@@ -77,13 +77,13 @@
 
 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
 
 

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 @@
         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 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)

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 @@
             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 @@
         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()