Merge pull request #923 from doronz88/refactor/os_interface

Organize OS-related functionality into classes

pymobiledevice3/__main__.py

@@ -6,10 +6,12 @@
 import coloredlogs
 
 from pymobiledevice3.exceptions import AccessDeniedError, ConnectionFailedToUsbmuxdError, DeprecationError, \
- DeveloperModeError, DeveloperModeIsNotEnabledError, DeviceHasPasscodeSetError, DeviceNotFoundError, InternalError, \
- InvalidServiceError, MessageNotSupportedError, MissingValueError, NoDeviceConnectedError, NoDeviceSelectedError, \
- NotEnoughDiskSpaceError, NotPairedError, PairingDialogResponsePendingError, PasswordRequiredError, \
- RSDRequiredError, SetProhibitedError, TunneldConnectionError, UserDeniedPairingError
+ DeveloperModeError, DeveloperModeIsNotEnabledError, DeviceHasPasscodeSetError, DeviceNotFoundError, \
+ FeatureNotSupportedError, InternalError, InvalidServiceError, MessageNotSupportedError, MissingValueError, \
+ NoDeviceConnectedError, NoDeviceSelectedError, NotEnoughDiskSpaceError, NotPairedError, OSNotSupportedError, \
+ PairingDialogResponsePendingError, PasswordRequiredError, RSDRequiredError, SetProhibitedError, \
+ TunneldConnectionError, UserDeniedPairingError
+from pymobiledevice3.osu.os_utils import get_os_utils
 
 coloredlogs.install(level=logging.INFO)
 
@@ -134,10 +136,7 @@ def main() -> None:
  except PasswordRequiredError:
  logger.error('Device is password protected. Please unlock and retry')
  except AccessDeniedError:
- if sys.platform == 'win32':
- logger.error('This command requires admin privileges. Consider retrying with "run-as administrator".')
- else:
- logger.error('This command requires root privileges. Consider retrying with "sudo".')
+ logger.error(get_os_utils().access_denied_error)
  except BrokenPipeError:
  traceback.print_exc()
  except TunneldConnectionError:
@@ -150,9 +149,18 @@ def main() -> None:
  logger.error('Not enough disk space')
  except RSDRequiredError:
  logger.error('The requested operation requires an RSD instance. For more information see:\n'
- 'https://github.com/doronz88/pymobiledevice3?tab=readme-ov-file#working-with-developer-tools-ios--170')
+ 'https://github.com/doronz88/pymobiledevice3?tab=readme-ov-file#working-with-developer-tools-ios'
+ '--170')
  except DeprecationError:
  logger.error('failed to query MobileGestalt, MobileGestalt deprecated (iOS >= 17.4).')
+ except OSNotSupportedError as e:
+ logger.error(
+ f'Unsupported OS - {e.os_name}. To add support, consider contributing at '
+ f'https://github.com/doronz88/pymobiledevice3.')
+ except FeatureNotSupportedError as e:
+ logger.error(
+ f'Missing implementation of `{e.feature}` on `{e.os_name}`. To add support, consider contributing at '
+ f'https://github.com/doronz88/pymobiledevice3.')
 
 
 if __name__ == '__main__':

pymobiledevice3/bonjour.py

@@ -1,17 +1,19 @@
 import asyncio
 import dataclasses
-import sys
 from socket import AF_INET, AF_INET6, inet_ntop
 from typing import List, Mapping, Optional
 
 from ifaddr import get_adapters
 from zeroconf import IPVersion, ServiceListener, ServiceStateChange, Zeroconf
 from zeroconf.asyncio import AsyncServiceBrowser, AsyncServiceInfo, AsyncZeroconf
 
+from pymobiledevice3.osu.os_utils import get_os_utils
+
 REMOTEPAIRING_SERVICE_NAMES = ['_remotepairing._tcp.local.']
 MOBDEV2_SERVICE_NAMES = ['_apple-mobdev2._tcp.local.']
 REMOTED_SERVICE_NAMES = ['_remoted._tcp.local.']
-DEFAULT_BONJOUR_TIMEOUT = 1 if sys.platform != 'win32' else 2 # On Windows, it takes longer to get the addresses
+OSUTILS = get_os_utils()
+DEFAULT_BONJOUR_TIMEOUT = OSUTILS.bonjour_timeout
 
 
 @dataclasses.dataclass
@@ -64,23 +66,6 @@ class BonjourQuery:
  listener: BonjourListener
 
 
-def get_ipv6_ips() -> List[str]:
- ips = []
- if sys.platform == 'win32':
- # TODO: verify on windows
- ips = [f'{adapter.ips[0].ip[0]}%{adapter.ips[0].ip[2]}' for adapter in get_adapters() if adapter.ips[0].is_IPv6]
- else:
- for adapter in get_adapters():
- for ip in adapter.ips:
- if not ip.is_IPv6:
- continue
- if ip.ip[0] in ('::1', 'fe80::1'):
- # skip localhost
- continue
- ips.append(f'{ip.ip[0]}%{adapter.nice_name}')
- return ips
-
-
 def query_bonjour(service_names: List[str], ip: str) -> BonjourQuery:
  aiozc = AsyncZeroconf(interfaces=[ip])
  listener = BonjourListener(ip)
@@ -106,7 +91,7 @@ async def browse(service_names: List[str], ips: List[str], timeout: float = DEFA
 
 
 async def browse_ipv6(service_names: List[str], timeout: float = DEFAULT_BONJOUR_TIMEOUT) -> List[BonjourAnswer]:
- return await browse(service_names, get_ipv6_ips(), timeout=timeout)
+ return await browse(service_names, OSUTILS.get_ipv6_ips(), timeout=timeout)
 
 
 async def browse_ipv4(service_names: List[str], timeout: float = DEFAULT_BONJOUR_TIMEOUT) -> List[BonjourAnswer]:

pymobiledevice3/cli/cli_common.py

@@ -2,7 +2,6 @@
 import json
 import logging
 import os
-import sys
 import uuid
 from functools import wraps
 from typing import Callable, List, Mapping, Optional, Tuple
@@ -18,12 +17,15 @@
 from pymobiledevice3.exceptions import AccessDeniedError, DeviceNotFoundError, NoDeviceConnectedError, \
  NoDeviceSelectedError
 from pymobiledevice3.lockdown import LockdownClient, create_using_usbmux
+from pymobiledevice3.osu.os_utils import get_os_utils
 from pymobiledevice3.remote.remote_service_discovery import RemoteServiceDiscoveryService
 from pymobiledevice3.tunneld import get_tunneld_devices
 from pymobiledevice3.usbmux import select_devices_by_connection_type
 
 USBMUX_OPTION_HELP = 'usbmuxd listener address (in the form of either /path/to/unix/socket OR HOST:PORT'
 COLORED_OUTPUT = True
+UDID_ENV_VAR = 'PYMOBILEDEVICE3_UDID'
+OSUTILS = get_os_utils()
 
 
 class RSDOption(Option):
@@ -108,42 +110,10 @@ def get_last_used_terminal_formatting(buf: str) -> str:
  return '\x1b' + buf.rsplit('\x1b', 1)[1].split('m')[0] + 'm'
 
 
-def wait_return() -> None:
- if sys.platform != 'win32':
- import signal
- print("Press Ctrl+C to send a SIGINT or use 'kill' command to send a SIGTERM")
- signal.sigwait([signal.SIGINT, signal.SIGTERM])
- else:
- input('Press ENTER to exit>')
-
-
-UDID_ENV_VAR = 'PYMOBILEDEVICE3_UDID'
-
-
-def is_admin_user() -> bool:
- """ Check if the current OS user is an Administrator or root.
-
- See: https://github.com/Preston-Landers/pyuac/blob/master/pyuac/admin.py
-
- :return: True if the current user is an 'Administrator', otherwise False.
- """
- if os.name == 'nt':
- import win32security
-
- try:
- admin_sid = win32security.CreateWellKnownSid(win32security.WinBuiltinAdministratorsSid, None)
- return win32security.CheckTokenMembership(None, admin_sid)
- except Exception:
- return False
- else:
- # Check for root on Posix
- return os.geteuid() == 0
-
-
 def sudo_required(func):
  @wraps(func)
  def wrapper(*args, **kwargs):
- if not is_admin_user():
+ if not OSUTILS.is_admin:
  raise AccessDeniedError()
  else:
  func(*args, **kwargs)

pymobiledevice3/cli/developer.py

@@ -19,11 +19,12 @@
 
 import pymobiledevice3
 from pymobiledevice3.cli.cli_common import BASED_INT, Command, RSDCommand, default_json_encoder, print_json, \
- user_requested_colored_output, wait_return
+ user_requested_colored_output
 from pymobiledevice3.exceptions import DeviceAlreadyInUseError, DvtDirListError, ExtractingStackshotError, \
  RSDRequiredError, UnrecognizedSelectorError
 from pymobiledevice3.lockdown import LockdownClient
 from pymobiledevice3.lockdown_service_provider import LockdownServiceProvider
+from pymobiledevice3.osu.os_utils import get_os_utils
 from pymobiledevice3.remote.core_device.app_service import AppServiceService
 from pymobiledevice3.remote.core_device.device_info import DeviceInfoService
 from pymobiledevice3.remote.remote_service_discovery import RemoteServiceDiscoveryService
@@ -52,6 +53,7 @@
 from pymobiledevice3.services.simulate_location import DtSimulateLocation
 from pymobiledevice3.tcp_forwarder import LockdownTcpForwarder
 
+OSUTILS = get_os_utils()
 BSC_SUBCLASS = 0x40c
 BSC_CLASS = 0x4
 VFS_AND_TRACES_SET = {0x03010000, 0x07ff0000}
@@ -813,7 +815,7 @@ def accessibility_settings_set(service_provider: LockdownClient, setting, value)
  """
  service = AccessibilityAudit(service_provider)
  service.set_setting(setting, eval(value))
- wait_return()
+ OSUTILS.wait_return()
 
 
 @accessibility.command('shell', cls=Command)
@@ -890,7 +892,7 @@ def condition_set(service_provider: LockdownClient, profile_identifier):
  """ set a specific condition """
  with DvtSecureSocketProxyService(lockdown=service_provider) as dvt:
  ConditionInducer(dvt).set(profile_identifier)
- wait_return()
+ OSUTILS.wait_return()
 
 
 @developer.command(cls=Command)
@@ -963,7 +965,7 @@ def check_in(service_provider: LockdownClient, hostname, force):
  with DtDeviceArbitration(service_provider) as device_arbitration:
  try:
  device_arbitration.check_in(hostname, force=force)
- wait_return()
+ OSUTILS.wait_return()
  except DeviceAlreadyInUseError as e:
  logger.error(e.message)
 
@@ -1009,7 +1011,7 @@ def dvt_simulate_location_set(service_provider: LockdownClient, latitude, longit
  """
  with DvtSecureSocketProxyService(service_provider) as dvt:
  LocationSimulation(dvt).set(latitude, longitude)
- wait_return()
+ OSUTILS.wait_return()
 
 
 @dvt_simulate_location.command('play', cls=Command)
@@ -1019,7 +1021,7 @@ def dvt_simulate_location_play(service_provider: LockdownClient, filename: str,
  """ play a .gpx file """
  with DvtSecureSocketProxyService(service_provider) as dvt:
  LocationSimulation(dvt).play_gpx_file(filename, disable_sleep=disable_sleep)
- wait_return()
+ OSUTILS.wait_return()
 
 
 @developer.group()

pymobiledevice3/cli/webinspector.py

@@ -21,11 +21,12 @@
 from pygments import formatters, highlight, lexers
 from pygments.styles import get_style_by_name
 
-from pymobiledevice3.cli.cli_common import Command, wait_return
+from pymobiledevice3.cli.cli_common import Command
 from pymobiledevice3.common import get_home_folder
 from pymobiledevice3.exceptions import InspectorEvaluateError, LaunchingApplicationError, \
  RemoteAutomationNotEnabledError, WebInspectorNotEnabledError, WirError
 from pymobiledevice3.lockdown import LockdownClient, create_using_usbmux
+from pymobiledevice3.osu.os_utils import get_os_utils
 from pymobiledevice3.services.web_protocol.cdp_server import app
 from pymobiledevice3.services.web_protocol.driver import By, Cookie, WebDriver
 from pymobiledevice3.services.web_protocol.inspector_session import InspectorSession
@@ -59,6 +60,7 @@
  'switch', 'synchronized', 'this', 'throw', 'throws', 'transient', 'true', 'try', 'typeof', 'var',
  'void', 'volatile', 'while', 'with', 'yield', ]
 
+OSUTILS = get_os_utils()
 logger = logging.getLogger(__name__)
 
 
@@ -153,7 +155,7 @@ def launch(service_provider: LockdownClient, url, timeout):
  driver.start_session()
  print('Getting URL')
  driver.get(url)
- wait_return()
+ OSUTILS.wait_return()
  session.stop_session()
  inspector.close()
 

pymobiledevice3/exceptions.py

@@ -13,6 +13,7 @@
  'NoDeviceSelectedError', 'MessageNotSupportedError', 'InvalidServiceError', 'InspectorEvaluateError',
  'LaunchingApplicationError', 'BadCommandError', 'BadDevError', 'ConnectionFailedError', 'CoreDeviceError',
  'AccessDeniedError', 'RSDRequiredError', 'SysdiagnoseTimeoutError', 'GetProhibitedError',
+ 'FeatureNotSupportedError', 'OSNotSupportedError'
 ]
 
 from typing import List, Optional
@@ -363,3 +364,22 @@ class RSDRequiredError(PyMobileDevice3Exception):
 class SysdiagnoseTimeoutError(PyMobileDevice3Exception, TimeoutError):
  """ Timeout collecting new sysdiagnose archive """
  pass
+
+
+class SupportError(PyMobileDevice3Exception):
+ def __init__(self, os_name):
+ self.os_name = os_name
+ super().__init__()
+
+
+class OSNotSupportedError(SupportError):
+ """ Operating system is not supported. """
+ pass
+
+
+class FeatureNotSupportedError(SupportError):
+ """ Feature has not been implemented for OS. """
+
+ def __init__(self, os_name, feature):
+ super().__init__(os_name)
+ self.feature = feature

pymobiledevice3/osu/os_utils.py

@@ -0,0 +1,81 @@
+import inspect
+import socket
+import sys
+from datetime import datetime
+from pathlib import Path
+from typing import List, Tuple
+
+from pymobiledevice3.exceptions import FeatureNotSupportedError, OSNotSupportedError
+
+DEFAULT_AFTER_IDLE_SEC = 3
+DEFAULT_INTERVAL_SEC = 3
+DEFAULT_MAX_FAILS = 3
+
+
+class OsUtils:
+ _instance = None
+ _os_name = None
+
+ @classmethod
+ def create(cls) -> 'OsUtils':
+ if cls._instance is None:
+ cls._os_name = sys.platform
+ if cls._os_name == 'win32':
+ from pymobiledevice3.osu.win_util import Win32
+ cls._instance = Win32()
+ elif cls._os_name == 'darwin':
+ from pymobiledevice3.osu.posix_util import Darwin
+ cls._instance = Darwin()
+ elif cls._os_name == 'linux':
+ from pymobiledevice3.osu.posix_util import Linux
+ cls._instance = Linux()
+ elif cls._os_name == 'cygwin':
+ from pymobiledevice3.osu.posix_util import Cygwin
+ cls._instance = Cygwin()
+ else:
+ raise OSNotSupportedError(cls._os_name)
+ return cls._instance
+
+ @property
+ def is_admin(self) -> bool:
+ raise FeatureNotSupportedError(self._os_name, inspect.currentframe().f_code.co_name)
+
+ @property
+ def usbmux_address(self) -> Tuple[str, int]:
+ raise FeatureNotSupportedError(self._os_name, inspect.currentframe().f_code.co_name)
+
+ @property
+ def bonjour_timeout(self) -> int:
+ raise FeatureNotSupportedError(self._os_name, inspect.currentframe().f_code.co_name)
+
+ @property
+ def loopback_header(self) -> bytes:
+ raise FeatureNotSupportedError(self._os_name, inspect.currentframe().f_code.co_name)
+
+ @property
+ def access_denied_error(self) -> str:
+ raise FeatureNotSupportedError(self._os_name, inspect.currentframe().f_code.co_name)
+
+ @property
+ def pair_record_path(self) -> Path:
+ raise FeatureNotSupportedError(self._os_name, inspect.currentframe().f_code.co_name)
+
+ def get_ipv6_ips(self) -> List[str]:
+ raise FeatureNotSupportedError(self._os_name, inspect.currentframe().f_code.co_name)
+
+ def set_keepalive(self, sock: socket.socket, after_idle_sec: int = DEFAULT_AFTER_IDLE_SEC,
+ interval_sec: int = DEFAULT_INTERVAL_SEC, max_fails: int = DEFAULT_MAX_FAILS) -> None:
+ raise FeatureNotSupportedError(self._os_name, inspect.currentframe().f_code.co_name)
+
+ def parse_timestamp(self, time_stamp) -> datetime:
+ raise FeatureNotSupportedError(self._os_name, inspect.currentframe().f_code.co_name)
+
+ def chown_to_non_sudo_if_needed(self, path: Path) -> None:
+ raise FeatureNotSupportedError(self._os_name, inspect.currentframe().f_code.co_name)
+
+ def wait_return(self):
+ raise FeatureNotSupportedError(self._os_name, inspect.currentframe().f_code.co_name)
+
+
+def get_os_utils() -> OsUtils:
+ return OsUtils.create()