From 360e52894dee55642bbdcea24e9282d288b8e3ad Mon Sep 17 00:00:00 2001 From: Chenli Date: Fri, 19 Jun 2026 12:25:38 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E9=80=82=E9=85=8D=20iOS=2018.2+=20?= =?UTF-8?q?=E5=B9=B6=E5=8D=87=E7=BA=A7=20pymobiledevice3=20=E8=87=B3=209.x?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 旧版 pymobiledevice3 2.46.1 无法连接 iOS 18.2 及以上设备（如 iOS 26）， 原因有二： 1. iOS 18.2+ 移除了 QUIC 隧道协议，必须改用 TCP 协议； 2. pymobiledevice3 9.x 将设备相关 API 全面改为 async，并移动/重命名了 大量符号（select_device、install_driver_if_required、 DvtSecureSocketProxyService 等已不存在）。 主要改动： - requirements.txt: pymobiledevice3 2.46.1 -> 9.27.0 - driver/connect.py: lockdown / AMFI 调用改为 async；用 get_core_device_tunnel_services + start_tunnel(protocol=TCP) 替代 已移除的 select_device/install_driver_if_required - main.py: 用 RemoteServiceDiscoveryService + DvtProvider + LocationSimulation 重写为 async（asyncio.run 驱动） - driver/location.py、run.py、init/init.py、init/tunnel.py: 跟随改为 async - run.py: 用 asyncio.sleep 取代忙等待循环 注：Python < 3.13 时 TCP 隧道依赖 sslpsk_pmd3 的 PSK-TLS 回退实现。 Co-Authored-By: Claude Opus 4.8 --- driver/connect.py | 51 +++++++++++++++++++++++++--------------------- driver/location.py | 10 ++++----- init/init.py | 8 ++++---- init/tunnel.py | 3 +-- main.py | 35 ++++++++++++++++++++----------- requirements.txt | 2 +- run.py | 14 +++++++------ 7 files changed, 69 insertions(+), 54 deletions(-) diff --git a/driver/connect.py b/driver/connect.py index b4be45d..411896a 100644 --- a/driver/connect.py +++ b/driver/connect.py @@ -1,55 +1,60 @@ -import logging import multiprocessing from pymobiledevice3.lockdown import create_using_usbmux, LockdownClient -from pymobiledevice3.cli.remote import install_driver_if_required -from pymobiledevice3.cli.remote import select_device, RemoteServiceDiscoveryService -from pymobiledevice3.cli.remote import start_tunnel -from pymobiledevice3.cli.remote import verify_tunnel_imports +from pymobiledevice3.remote.common import TunnelProtocol +from pymobiledevice3.remote.module_imports import start_tunnel, verify_tunnel_imports +from pymobiledevice3.remote.tunnel_service import get_core_device_tunnel_services from pymobiledevice3.services.amfi import AmfiService from pymobiledevice3.exceptions import NoDeviceConnectedError -def get_usbmux_lockdownclient(): +async def get_usbmux_lockdownclient(): while True: try: - lockdown = create_using_usbmux() + lockdown = await create_using_usbmux() except NoDeviceConnectedError: print("请连接设备后按回车...") input() else: break while True: - lockdown = create_using_usbmux() + lockdown = await create_using_usbmux() if lockdown.all_values.get("PasswordProtected"): print("请解锁设备后按回车...") input() else: break - return create_using_usbmux() + return await create_using_usbmux() def get_version(lockdown: LockdownClient): return lockdown.all_values.get("ProductVersion") -def get_developer_mode_status(lockdown: LockdownClient): - return lockdown.developer_mode_status +async def get_developer_mode_status(lockdown: LockdownClient): + return await lockdown.get_developer_mode_status() -def reveal_developer_mode(lockdown: LockdownClient): - AmfiService(lockdown).create_amfi_show_override_path_file() +async def reveal_developer_mode(lockdown: LockdownClient): + await AmfiService(lockdown).reveal_developer_mode_option_in_ui() -def enable_developer_mode(lockdown: LockdownClient): - AmfiService(lockdown).enable_developer_mode() +async def enable_developer_mode(lockdown: LockdownClient): + await AmfiService(lockdown).enable_developer_mode() -def get_serverrsd(): - install_driver_if_required() + +async def tunnel(queue: multiprocessing.Queue): if not verify_tunnel_imports(): exit(1) - return select_device(None) - -async def tunnel(rsd: RemoteServiceDiscoveryService, queue: multiprocessing.Queue): - async with start_tunnel(rsd, None) as tunnel_result: - queue.put((tunnel_result.address, tunnel_result.port)) - await tunnel_result.client.wait_closed() + services = await get_core_device_tunnel_services() + if not services: + raise NoDeviceConnectedError() + service = services[0] + + try: + # iOS 18.2+ removed QUIC tunnel support, so TCP must be used. On + # Python < 3.13 this relies on the sslpsk_pmd3 PSK-TLS fallback. + async with start_tunnel(service, protocol=TunnelProtocol.TCP) as tunnel_result: + queue.put((tunnel_result.address, tunnel_result.port)) + await tunnel_result.client.wait_closed() + finally: + await service.close() diff --git a/driver/location.py b/driver/location.py index 6a602fc..9500862 100644 --- a/driver/location.py +++ b/driver/location.py @@ -1,7 +1,5 @@ -from pymobiledevice3.cli.developer import LocationSimulation +async def set_location(location_simulation, lat: float, lng: float): + await location_simulation.set(lat, lng) -def set_location(dvt, lat: float, lng: float): - LocationSimulation(dvt).set(lat, lng) - -def clear_location(dvt): - LocationSimulation(dvt).clear() +async def clear_location(location_simulation): + await location_simulation.clear() diff --git a/init/init.py b/init/init.py index de453bf..e409349 100644 --- a/init/init.py +++ b/init/init.py @@ -4,7 +4,7 @@ from driver import connect -def init(): +async def init(): # check if root on mac or Administrator on windows if sys.platform == "win32": if not ctypes.windll.shell32.IsUserAnAdmin(): @@ -19,7 +19,7 @@ def init(): sys.exit(1) # get lockdown client - lockdown = connect.get_usbmux_lockdownclient() + lockdown = await connect.get_usbmux_lockdownclient() # check version version = connect.get_version(lockdown) @@ -29,8 +29,8 @@ def init(): sys.exit(1) # check developer mode status - developer_mode_status = connect.get_developer_mode_status(lockdown) + developer_mode_status = await connect.get_developer_mode_status(lockdown) if not developer_mode_status: - connect.reveal_developer_mode(lockdown) + await connect.reveal_developer_mode(lockdown) print("您未开启开发者模式，请打开设备的 设置-隐私与安全性-开发者模式 来开启，开启后需要重启并输入密码，完成后再次运行此程序") sys.exit(1) \ No newline at end of file diff --git a/init/tunnel.py b/init/tunnel.py index 5119d90..db0e776 100644 --- a/init/tunnel.py +++ b/init/tunnel.py @@ -4,8 +4,7 @@ from driver import connect def tunnel_proc(queue: multiprocessing.Queue): - server_rsd = connect.get_serverrsd() - asyncio.run(connect.tunnel(server_rsd, queue)) + asyncio.run(connect.tunnel(queue)) def tunnel(): diff --git a/main.py b/main.py index cb273db..3449ec8 100644 --- a/main.py +++ b/main.py @@ -1,12 +1,14 @@ import signal import logging +import asyncio import coloredlogs import os from driver import location -from pymobiledevice3.cli.remote import RemoteServiceDiscoveryService -from pymobiledevice3.cli.developer import DvtSecureSocketProxyService +from pymobiledevice3.remote.remote_service_discovery import RemoteServiceDiscoveryService +from pymobiledevice3.services.dvt.instruments.dvt_provider import DvtProvider +from pymobiledevice3.services.dvt.instruments.location_simulation import LocationSimulation from init import init from init import tunnel @@ -34,7 +36,7 @@ -def main(): +async def amain(): # set level logger = logging.getLogger(__name__) coloredlogs.install(level=logging.INFO) @@ -43,7 +45,7 @@ def main(): logger.setLevel(logging.DEBUG) coloredlogs.install(level=logging.DEBUG) - init.init() + await init.init() logger.info("init done") # start the tunnel in another process @@ -59,22 +61,25 @@ def main(): loc = route.get_route() logger.info(f"got route from {config.config.routeConfig}") - - with RemoteServiceDiscoveryService((address, port)) as rsd: - with DvtSecureSocketProxyService(rsd) as dvt: + rsd = RemoteServiceDiscoveryService((address, port)) + await rsd.connect() + try: + async with DvtProvider(rsd) as dvt, LocationSimulation(dvt) as location_simulation: try: print(f"已开始模拟跑步，速度大约为 {config.config.v} m/s") print("会无限循环，按 Ctrl+C 退出") print("请勿直接关闭窗口，否则无法还原正常定位") - run.run(dvt, loc, config.config.v) - except KeyboardInterrupt: + await run.run(location_simulation, loc, config.config.v) + except (KeyboardInterrupt, asyncio.CancelledError): logger.debug("get KeyboardInterrupt (inner)") logger.debug(f"Is process alive? {process.is_alive()}") finally: logger.debug(f"Is process alive? {process.is_alive()}") logger.debug("Start to clear location") - location.clear_location(dvt) + await location.clear_location(location_simulation) logger.info("Location cleared") + finally: + await rsd.close() except KeyboardInterrupt: @@ -86,8 +91,14 @@ def main(): process.terminate() logger.info("tunnel process terminated") print("Bye") - - + +def main(): + try: + asyncio.run(amain()) + except KeyboardInterrupt: + pass + + if __name__ == "__main__": main() \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index a35c754..102192d 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,3 +1,3 @@ -pymobiledevice3==2.46.1 +pymobiledevice3==9.27.0 PyYAML==6.0.1 geopy==2.4.1 diff --git a/run.py b/run.py index 917299e..56606ab 100644 --- a/run.py +++ b/run.py @@ -7,6 +7,7 @@ import math import time import random +import asyncio from geopy.distance import geodesic @@ -134,7 +135,7 @@ def fixLockT(loc: list, v, dt): t += dt return fixedLoc -def run1(dvt, loc: list, v, dt=0.2): +async def run1(location_simulation, loc: list, v, dt=0.2): fixedLoc = fixLockT(loc, v, dt) nList = (5, 6, 7, 8, 9) n = nList[random.randint(0, len(nList)-1)] @@ -142,14 +143,15 @@ def run1(dvt, loc: list, v, dt=0.2): clock = time.time() for i in fixedLoc: # utils.setLoc(bd09Towgs84(i)) - location.set_location(dvt, **bd09Towgs84(i)) - while time.time()-clock < dt: - pass + await location.set_location(location_simulation, **bd09Towgs84(i)) + remaining = dt - (time.time() - clock) + if remaining > 0: + await asyncio.sleep(remaining) clock = time.time() -def run(dvt, loc: list, v, d=15): +async def run(location_simulation, loc: list, v, d=15): random.seed(time.time()) while True: vRand = 1000/(1000/v-(2*random.random()-1)*d) - run1(dvt, loc, vRand) + await run1(location_simulation, loc, vRand) print("跑完一圈了") \ No newline at end of file