diff --git a/.github/workflows/pull-request.yml b/.github/workflows/pull-request.yml index b1c1098e..534165c8 100644 --- a/.github/workflows/pull-request.yml +++ b/.github/workflows/pull-request.yml @@ -23,7 +23,7 @@ jobs: strategy: matrix: os: [ubuntu-latest, windows-latest, macos-latest] - python-version: [ 3.8, 3.9, "3.10", 3.11, 3.12 ] + python-version: [ 3.9, "3.10", 3.11, 3.12 ] steps: - uses: actions/checkout@v3 @@ -106,9 +106,6 @@ jobs: strategy: matrix: include: - - os: macos-latest - arch: arm64 - python-version: "3.8" - os: macos-latest arch: arm64 python-version: "3.9" @@ -122,9 +119,6 @@ jobs: arch: arm64 python-version: "3.12" - - os: macos-latest - arch: x86_64 - python-version: "3.8" - os: macos-latest arch: x86_64 python-version: "3.9" diff --git a/src/rpcclient/pyproject.toml b/src/rpcclient/pyproject.toml index ab20809a..2cd34c5e 100644 --- a/src/rpcclient/pyproject.toml +++ b/src/rpcclient/pyproject.toml @@ -2,7 +2,7 @@ name = "rpcclient" description = "rpcclient for connecting with the rpcserver" readme = "README.md" -requires-python = ">=3.8" +requires-python = ">=3.9" license = { text = "GNU GENERAL PUBLIC LICENSE Version 3, 29 June 2007" } keywords = ["ios", "macos", "linux", "automation", "remote-shell", "remote-control", "ipython"] authors = [ @@ -19,7 +19,6 @@ classifiers = [ "Development Status :: 5 - Production/Stable", "License :: OSI Approved :: GNU General Public License v3 or later (GPLv3+)", "Programming Language :: Python :: 3", - "Programming Language :: Python :: 3.8", "Programming Language :: Python :: 3.9", "Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.11", diff --git a/src/rpcclient/rpcclient/client.py b/src/rpcclient/rpcclient/client.py index 13493a0a..08345a9b 100644 --- a/src/rpcclient/rpcclient/client.py +++ b/src/rpcclient/rpcclient/client.py @@ -12,7 +12,7 @@ from collections import namedtuple from pathlib import Path from select import select -from typing import Any +from typing import Any, Optional import IPython from traitlets.config import Config @@ -164,7 +164,7 @@ def dlsym(self, lib: int, symbol_name: str): response = self._sock.send_recv(command) return response.ptr - def call(self, address: int, argv: typing.List[int] = None, return_float64=False, return_float32=False, + def call(self, address: int, argv: list[int] = None, return_float64=False, return_float32=False, return_raw=False, va_list_index: int = 0xffff) -> typing.Union[float, Symbol, Any]: """ call a remote function and retrieve its return value as Symbol object """ args = [] @@ -245,7 +245,7 @@ def listdir(self, filename: str): stat=stat)) return entries - def spawn(self, argv: typing.List[str] = None, envp: typing.List[str] = None, stdin: io_or_str = sys.stdin, + def spawn(self, argv: list[str] = None, envp: list[str] = None, stdin: io_or_str = sys.stdin, stdout=sys.stdout, raw_tty=False, background=False) -> SpawnResult: """ spawn a new process and forward its stdin, stdout & stderr @@ -317,7 +317,7 @@ def last_error(self): return f'[{self.errno}] {err_str}' @property - def environ(self) -> typing.List[str]: + def environ(self) -> list[str]: result = [] environ = self.symbols.environ[0] i = 0 @@ -363,7 +363,7 @@ def freeing(self, symbol): if symbol: self.symbols.free(symbol) - def interactive(self, additional_namespace: typing.Mapping = None): + def interactive(self, additional_namespace: Optional[dict] = None): """ Start an interactive shell """ sys.argv = ['a'] c = Config() @@ -451,7 +451,7 @@ def reconnect(self): # new clients are handled in new processes so all symbols may reside in different addresses self._init_process_specific() - def _execute(self, argv: typing.List[str], envp: typing.List[str], background=False) -> int: + def _execute(self, argv: list[str], envp: list[str], background=False) -> int: command = CmdExec(background=background, argv=argv, envp=envp) try: return self._sock.send_recv(command).pid diff --git a/src/rpcclient/rpcclient/darwin/cfpreferences.py b/src/rpcclient/rpcclient/darwin/cfpreferences.py index 5ba809f8..9fb482e5 100644 --- a/src/rpcclient/rpcclient/darwin/cfpreferences.py +++ b/src/rpcclient/rpcclient/darwin/cfpreferences.py @@ -23,7 +23,7 @@ def __init__(self, client): self._client = client def get_keys(self, application_id: str, username: str = kCFPreferencesCurrentUser, - hostname: str = kCFPreferencesCurrentHost) -> typing.Optional[typing.List[str]]: + hostname: str = kCFPreferencesCurrentHost) -> typing.Optional[list[str]]: """ wrapper for CFPreferencesCopyKeyList """ application_id = self._client.cf(application_id) username = self._client.cf(username) @@ -43,7 +43,7 @@ def get_value(self, key: str, application_id: str, username: str = kCFPreference return self._client.symbols.CFPreferencesCopyValue(key, application_id, username, hostname).py() def get_dict(self, application_id: str, username: str = kCFPreferencesCurrentUser, - hostname: str = kCFPreferencesCurrentHost) -> typing.Optional[typing.Mapping]: + hostname: str = kCFPreferencesCurrentHost) -> typing.Optional[dict]: """ get a dictionary representation of given preference """ result = {} key_list = self.get_keys(application_id, username, hostname) @@ -67,14 +67,14 @@ def remove(self, key: str, application_id: str, username: str = kCFPreferencesCu self._client.cf(application_id), self._client.cf(username), self._client.cf(hostname)) - def set_dict(self, d: typing.Mapping, application_id: str, username: str = kCFPreferencesCurrentUser, + def set_dict(self, d: dict, application_id: str, username: str = kCFPreferencesCurrentUser, hostname: str = kCFPreferencesCurrentHost): """ set entire preference dictionary (erase first if exists) """ with suppress(NoSuchPreferenceError): self.clear(application_id, username, hostname) self.update_dict(d, application_id, username, hostname) - def update_dict(self, d: typing.Mapping, application_id: str, username: str = kCFPreferencesCurrentUser, + def update_dict(self, d: dict, application_id: str, username: str = kCFPreferencesCurrentUser, hostname: str = kCFPreferencesCurrentHost): """ update preference dictionary """ for k, v in d.items(): diff --git a/src/rpcclient/rpcclient/darwin/client.py b/src/rpcclient/rpcclient/darwin/client.py index 5556b05f..2b5037bd 100644 --- a/src/rpcclient/rpcclient/darwin/client.py +++ b/src/rpcclient/rpcclient/darwin/client.py @@ -7,7 +7,6 @@ from collections import namedtuple from dataclasses import dataclass from functools import lru_cache -from typing import Mapping from cached_property import cached_property from tqdm import tqdm @@ -75,7 +74,7 @@ def __init__(self, sock, sysname: str, arch, create_socket_cb: typing.Callable): super().__init__(sock, sysname, arch, create_socket_cb, dlsym_global_handle=RTLD_GLOBAL) def _init_process_specific(self): - super(DarwinClient, self)._init_process_specific() + super()._init_process_specific() if 0 == self.dlopen('/System/Library/Frameworks/CoreFoundation.framework/CoreFoundation', RTLD_NOW): raise MissingLibraryError('failed to load CoreFoundation') @@ -100,7 +99,7 @@ def _init_process_specific(self): self._NSPropertyListSerialization = self.symbols.objc_getClass('NSPropertyListSerialization') self._CFNullTypeID = self.symbols.CFNullGetTypeID() - def interactive(self, additional_namespace: typing.Mapping = None): + def interactive(self, additional_namespace: typing.Optional[dict] = None): if additional_namespace is None: additional_namespace = {} additional_namespace['CFSTR'] = self.cf @@ -119,7 +118,7 @@ def errno(self, value: int) -> None: p_error[0] = value @property - def images(self) -> typing.List[DyldImage]: + def images(self) -> list[DyldImage]: m = [] for i in range(self.symbols._dyld_image_count()): module_name = self.symbols._dyld_get_image_name(i).peek_str() @@ -140,19 +139,19 @@ def is_idevice(self): return self.uname.machine.startswith('i') @property - def roots(self) -> typing.List[str]: + def roots(self) -> list[str]: """ get a list of all accessible darwin roots when used for lookup of files/preferences/... """ return ['/', '/var/root'] - def showobject(self, object_address: Symbol) -> Mapping: + def showobject(self, object_address: Symbol) -> dict: response = self._sock.send_recv(CmdShowObject(address=object_address)) return json.loads(response.description) - def showclass(self, class_address: Symbol) -> Mapping: + def showclass(self, class_address: Symbol) -> dict: response = self._sock.send_recv(CmdShowClass(address=class_address)) return json.loads(response.description) - def get_class_list(self) -> typing.Mapping[str, objective_c_class.Class]: + def get_class_list(self) -> dict[str, objective_c_class.Class]: result = {} command = CmdGetClassList() response = self._sock.send_recv(command) diff --git a/src/rpcclient/rpcclient/darwin/common.py b/src/rpcclient/rpcclient/darwin/common.py index 06b76fe9..8e6d22a3 100644 --- a/src/rpcclient/rpcclient/darwin/common.py +++ b/src/rpcclient/rpcclient/darwin/common.py @@ -1,5 +1,5 @@ from datetime import datetime -from typing import Any, List, Mapping, Tuple, Union +from typing import Any, Union CfSerializable = Union[ - Mapping[str, Any], List, Tuple[Any, ...], str, bool, float, bytes, datetime, None] + tuple[str, Any], list, tuple[Any, ...], str, bool, float, bytes, datetime, None] diff --git a/src/rpcclient/rpcclient/darwin/crash_reports.py b/src/rpcclient/rpcclient/darwin/crash_reports.py index e246dd8b..74a69e2c 100644 --- a/src/rpcclient/rpcclient/darwin/crash_reports.py +++ b/src/rpcclient/rpcclient/darwin/crash_reports.py @@ -1,5 +1,4 @@ from pathlib import Path -from typing import List from pycrashreport.crash_report import CrashReportBase, get_crash_report_from_buf @@ -21,7 +20,7 @@ def set_symbolicated(self, enabled: bool = True): # bugfix: at some point, this setting was moved to "com.apple.osanalytics" bundle identifier self._client.preferences.cf.set('SymbolicateCrashes', enabled, 'com.apple.osanalytics', 'root') - def list(self, prefixed='') -> List[CrashReportBase]: + def list(self, prefixed='') -> list[CrashReportBase]: """ get a list of all crash reports as CrashReport parsed objects """ result = [] for root in self._client.roots: diff --git a/src/rpcclient/rpcclient/darwin/darwin_lief.py b/src/rpcclient/rpcclient/darwin/darwin_lief.py index 5ecc2e8e..b6f08c53 100644 --- a/src/rpcclient/rpcclient/darwin/darwin_lief.py +++ b/src/rpcclient/rpcclient/darwin/darwin_lief.py @@ -1,6 +1,5 @@ import plistlib import struct -from typing import Mapping import lief from parameter_decorators import path_to_str @@ -12,7 +11,7 @@ class DarwinLief(Lief): @path_to_str('path') - def get_entitlements(self, path: str) -> Mapping: + def get_entitlements(self, path: str) -> dict: with self._client.fs.open(path, 'r') as f: buf = f.read() parsed = lief.parse(buf) diff --git a/src/rpcclient/rpcclient/darwin/fs.py b/src/rpcclient/rpcclient/darwin/fs.py index 272c25ff..bd331e12 100644 --- a/src/rpcclient/rpcclient/darwin/fs.py +++ b/src/rpcclient/rpcclient/darwin/fs.py @@ -1,5 +1,3 @@ -from typing import List, Mapping - from parameter_decorators import path_to_str from rpcclient.darwin.structs import stat64, statfs64 @@ -52,7 +50,7 @@ def removexattr(self, path: str, name: str): self._client.raise_errno_exception(f'failed to removexattr(): {path}') @path_to_str('path') - def listxattr(self, path: str) -> List[str]: + def listxattr(self, path: str) -> list[str]: """ list extended attribute names """ max_buf_len = 1024 with self._client.safe_malloc(max_buf_len) as xattributes_names: @@ -72,7 +70,7 @@ def getxattr(self, path: str, name: str) -> bytes: return value.peek(count) @path_to_str('path') - def dictxattr(self, path: str) -> Mapping[str, bytes]: + def dictxattr(self, path: str) -> dict[str, bytes]: """ get a dictionary of all extended attributes """ result = {} for k in self.listxattr(path): diff --git a/src/rpcclient/rpcclient/darwin/ioregistry.py b/src/rpcclient/rpcclient/darwin/ioregistry.py index 3eac09b5..7790a9c6 100644 --- a/src/rpcclient/rpcclient/darwin/ioregistry.py +++ b/src/rpcclient/rpcclient/darwin/ioregistry.py @@ -1,5 +1,3 @@ -from typing import Mapping - from rpcclient.allocated import Allocated from rpcclient.darwin.consts import MACH_PORT_NULL, kCFAllocatorDefault, kIOServicePlane from rpcclient.darwin.structs import io_name_t, io_object_t, mach_port_t @@ -22,7 +20,7 @@ def name(self) -> str: return name.peek_str() @property - def properties(self) -> Mapping: + def properties(self) -> dict: with self._client.safe_malloc(8) as p_properties: if self._client.symbols.IORegistryEntryCreateCFProperties(self._service, p_properties, kCFAllocatorDefault, 0): @@ -42,7 +40,7 @@ def __iter__(self): s = IOService(self._client, child) yield s - def set(self, properties: Mapping): + def set(self, properties: dict): self._client.symbols.IORegistryEntrySetCFProperties(self._service, self._client.cf(properties)) def get(self, key: str): @@ -58,7 +56,7 @@ def __repr__(self): class BacklightControlService(IOService): @property - def display_parameters(self) -> Mapping: + def display_parameters(self) -> dict: return self.get('IODisplayParameters') @property diff --git a/src/rpcclient/rpcclient/darwin/keychain.py b/src/rpcclient/rpcclient/darwin/keychain.py index 8f9881d7..f5009b92 100644 --- a/src/rpcclient/rpcclient/darwin/keychain.py +++ b/src/rpcclient/rpcclient/darwin/keychain.py @@ -1,5 +1,4 @@ import logging -from typing import List, Mapping from rpcclient.exceptions import BadReturnValueError, RpcPermissionError from rpcclient.symbol import Symbol @@ -24,25 +23,25 @@ def add_internet_password(self, account: str, server: str, password: str): if err != 0: raise BadReturnValueError(f'SecItemAdd() returned: {err}') - def query_apple_share_passwords(self) -> List[Mapping]: + def query_apple_share_passwords(self) -> list[dict]: return self._query(self._client.symbols.kSecClassAppleSharePassword) - def query_internet_passwords(self) -> List[Mapping]: + def query_internet_passwords(self) -> list[dict]: return self._query(self._client.symbols.kSecClassInternetPassword) - def query_generic_passwords(self) -> List[Mapping]: + def query_generic_passwords(self) -> list[dict]: return self._query(self._client.symbols.kSecClassGenericPassword) - def query_identities(self) -> List[Mapping]: + def query_identities(self) -> list[dict]: return self._query(self._client.symbols.kSecClassIdentity) - def query_certificates(self) -> List[Mapping]: + def query_certificates(self) -> list[dict]: return self._query(self._client.symbols.kSecClassCertificate) - def query_keys(self) -> List[Mapping]: + def query_keys(self) -> list[dict]: return self._query(self._client.symbols.kSecClassKey) - def _query(self, class_type: Symbol) -> List[Mapping]: + def _query(self, class_type: Symbol) -> list[dict]: with self._client.safe_malloc(8) as p_result: p_result[0] = 0 diff --git a/src/rpcclient/rpcclient/darwin/location.py b/src/rpcclient/rpcclient/darwin/location.py index 27353ebf..eff54f16 100644 --- a/src/rpcclient/rpcclient/darwin/location.py +++ b/src/rpcclient/rpcclient/darwin/location.py @@ -1,5 +1,5 @@ from enum import Enum -from typing import Mapping, Optional +from typing import Optional from rpcclient.exceptions import MissingLibraryError, RpcPermissionError from rpcclient.structs.consts import RTLD_NOW @@ -61,7 +61,7 @@ def authorization_status(self) -> CLAuthorizationStatus: return CLAuthorizationStatus.from_value(self._location_manager.objc_call('authorizationStatus')) @property - def last_sample(self) -> Optional[Mapping]: + def last_sample(self) -> Optional[dict]: """ last taken location sample (or None if there isn't any) """ location = self._location_manager.objc_call('location') if not location: diff --git a/src/rpcclient/rpcclient/darwin/media.py b/src/rpcclient/rpcclient/darwin/media.py index 82fc44eb..adda9e2e 100644 --- a/src/rpcclient/rpcclient/darwin/media.py +++ b/src/rpcclient/rpcclient/darwin/media.py @@ -1,6 +1,5 @@ import struct from enum import Enum -from typing import List from parameter_decorators import path_to_str @@ -158,7 +157,7 @@ def record_permission(self): return struct.pack(' List[str]: + def available_categories(self) -> list[str]: return self._session.objc_call('availableCategories').py() diff --git a/src/rpcclient/rpcclient/darwin/network.py b/src/rpcclient/rpcclient/darwin/network.py index a76a0948..9567430a 100644 --- a/src/rpcclient/rpcclient/darwin/network.py +++ b/src/rpcclient/rpcclient/darwin/network.py @@ -1,5 +1,4 @@ import sqlite3 -from typing import Mapping from rpcclient.network import Network from rpcclient.structs.consts import SIGKILL @@ -15,7 +14,7 @@ def __init__(self, client): super().__init__(client) @property - def proxy_settings(self) -> Mapping: + def proxy_settings(self) -> dict: return self._client.symbols.CFNetworkCopySystemProxySettings().py() def set_http_proxy(self, ip: str, port: int) -> None: diff --git a/src/rpcclient/rpcclient/darwin/objc.py b/src/rpcclient/rpcclient/darwin/objc.py index 6d65b9b7..ae6f9c76 100644 --- a/src/rpcclient/rpcclient/darwin/objc.py +++ b/src/rpcclient/rpcclient/darwin/objc.py @@ -1,6 +1,5 @@ from collections import namedtuple from dataclasses import dataclass, field -from typing import Mapping from objc_types_decoder.decode import decode as decode_type from objc_types_decoder.decode import decode_with_tail @@ -49,7 +48,7 @@ class Method: args_types: list = field(compare=False) @staticmethod - def from_data(data: Mapping, client): + def from_data(data: dict, client): """ Create Method object from raw data. :param data: Data as loaded from get_objectivec_symbol_data.m. diff --git a/src/rpcclient/rpcclient/darwin/objective_c_class.py b/src/rpcclient/rpcclient/darwin/objective_c_class.py index a048572e..6efcba52 100644 --- a/src/rpcclient/rpcclient/darwin/objective_c_class.py +++ b/src/rpcclient/rpcclient/darwin/objective_c_class.py @@ -1,7 +1,7 @@ from collections import namedtuple from functools import partial from pathlib import Path -from typing import Mapping, Optional +from typing import Optional from pygments import highlight from pygments.formatters import TerminalTrueColorFormatter @@ -19,7 +19,7 @@ class Class: Wrapper for ObjectiveC Class object. """ - def __init__(self, client, class_object=0, class_data: Mapping = None, lazy=False): + def __init__(self, client, class_object=0, class_data: Optional[dict] = None, lazy=False): """ :param rpcclient.darwin.client.DarwinClient client: Darwin client. :param rpcclient.darwin.objective_c_symbol.Symbol class_object: @@ -126,7 +126,7 @@ def iter_supers(self): yield sup sup = sup.super - def _load_class_data(self, data: Mapping): + def _load_class_data(self, data: dict): self._class_object = self._client.symbol(data['address']) self.super = Class(self._client, data['super']) if data['super'] else None self.name = data['name'] @@ -162,7 +162,7 @@ def __dir__(self): if method.is_class: result.add(method.name.replace(':', '_')) - result.update(list(super(Class, self).__dir__())) + result.update(list(super().__dir__())) return list(result) def __str__(self): diff --git a/src/rpcclient/rpcclient/darwin/objective_c_symbol.py b/src/rpcclient/rpcclient/darwin/objective_c_symbol.py index c4a0cfce..13c7e143 100644 --- a/src/rpcclient/rpcclient/darwin/objective_c_symbol.py +++ b/src/rpcclient/rpcclient/darwin/objective_c_symbol.py @@ -9,6 +9,7 @@ from pygments.lexers import ObjectiveCLexer from rpcclient.darwin import objc +from rpcclient.darwin.objc import Method from rpcclient.darwin.objective_c_class import Class from rpcclient.darwin.symbol import DarwinSymbol from rpcclient.exceptions import RpcClientException @@ -98,8 +99,18 @@ def objc_call(self, selector: str, *params, **kwargs): :param params: Additional parameters. :return: ObjectiveCSymbol when return type is an objc symbol. """ - symbol = super(ObjectiveCSymbol, self).objc_call(selector, *params, **kwargs) - return symbol.objc_symbol if self._client.is_objc_type(symbol) else symbol + symbol = super().objc_call(selector, *params, **kwargs) + try: + is_objc_type = self.get_method(selector).return_type == 'id' + except AttributeError: + is_objc_type = False + return symbol.objc_symbol if is_objc_type else symbol + + def get_method(self, name: str) -> Method: + for method in self.methods: + if method.name == name: + return method + raise AttributeError(f'Method "{name}" does not exist') def _set_ivar(self, name, value): try: @@ -182,12 +193,12 @@ def __dir__(self): for method in sup.methods: result.add(method.name.replace(':', '_')) - result.update(list(super(ObjectiveCSymbol, self).__dir__())) + result.update(list(super().__dir__())) return list(result) def __getitem__(self, item): if isinstance(item, int): - return super(ObjectiveCSymbol, self).__getitem__(item) + return super().__getitem__(item) # Ivars for ivar in self.ivars: @@ -218,7 +229,7 @@ def __getattr__(self, item: str): def __setitem__(self, key, value): if isinstance(key, int): - super(ObjectiveCSymbol, self).__setitem__(key, value) + super().__setitem__(key, value) return with suppress(SettingIvarError): @@ -233,7 +244,7 @@ def __setattr__(self, key, value): try: self._set_ivar(key, value) except SettingIvarError: - super(ObjectiveCSymbol, self).__setattr__(key, value) + super().__setattr__(key, value) def __str__(self): return self._to_str(False) diff --git a/src/rpcclient/rpcclient/darwin/power.py b/src/rpcclient/rpcclient/darwin/power.py index 576b28bf..c69a69b2 100644 --- a/src/rpcclient/rpcclient/darwin/power.py +++ b/src/rpcclient/rpcclient/darwin/power.py @@ -1,5 +1,4 @@ import logging -from typing import List, Mapping from rpcclient.darwin.consts import IOPMUserActiveType from rpcclient.exceptions import BadReturnValueError @@ -54,7 +53,7 @@ def create_assertion(self, name: str, type_: str, reason: str = None) -> PowerAs raise BadReturnValueError('IOPMAssertionCreateWithProperties() failed') return PowerAssertion(self._client, p_assertion_id[0]) - def copy_assertions_by_process(self) -> Mapping[int, Mapping]: + def copy_assertions_by_process(self) -> dict[int, dict]: """ Returns a dictionary listing all assertions, grouped by their owning process """ with self._client.safe_malloc(8) as p_assertions: if self._client.symbols.IOPMCopyAssertionsByProcess(p_assertions) != 0: @@ -72,11 +71,11 @@ def copy_assertions_by_process(self) -> Mapping[int, Mapping]: result[pid] = assertions.objc_call('objectForKey:', pid_object).py() return result - def copy_scheduled_power_events(self) -> List[Mapping]: + def copy_scheduled_power_events(self) -> list[dict]: """ List all scheduled system power events """ return self._client.symbols.IOPMCopyScheduledPowerEvents().py() - def copy_assertions_status(self) -> Mapping[str, int]: + def copy_assertions_status(self) -> dict[str, int]: """ Returns a list of available assertions and their system-wide levels """ with self._client.safe_malloc(8) as result: if 0 != self._client.symbols.IOPMCopyAssertionsStatus(result): diff --git a/src/rpcclient/rpcclient/darwin/processes.py b/src/rpcclient/rpcclient/darwin/processes.py index e8ab6553..f5e55c8e 100644 --- a/src/rpcclient/rpcclient/darwin/processes.py +++ b/src/rpcclient/rpcclient/darwin/processes.py @@ -6,9 +6,10 @@ import struct import time from collections import namedtuple +from collections.abc import Generator from datetime import datetime from pathlib import Path -from typing import Generator, List, Mapping, Optional +from typing import Optional from cached_property import cached_property from construct import Array, Container, Int32ul @@ -139,7 +140,7 @@ def thread_id(self) -> int: def get_state(self): raise NotImplementedError() - def set_state(self, state: Mapping): + def set_state(self, state: dict): raise NotImplementedError() def resume(self): @@ -162,7 +163,7 @@ def get_state(self): raise BadReturnValueError('thread_get_state() failed') return x86_thread_state64_t.parse_stream(p_state) - def set_state(self, state: Mapping): + def set_state(self, state: dict) -> None: if self._client.symbols.thread_set_state(self._thread_id, x86_THREAD_STATE64, x86_thread_state64_t.build(state), x86_thread_state64_t.sizeof() // Int32ul.sizeof()): @@ -179,7 +180,7 @@ def get_state(self): raise BadReturnValueError('thread_get_state() failed') return arm_thread_state64_t.parse_stream(p_state) - def set_state(self, state: Mapping): + def set_state(self, state: dict) -> None: if self._client.symbols.thread_set_state(self._thread_id, ARMThreadFlavors.ARM_THREAD_STATE64, arm_thread_state64_t.build(state), ARM_THREAD_STATE64_COUNT): @@ -237,7 +238,7 @@ class Backtrace: pid: int thread_id: int dispatch_queue_serial_num: int - frames: List[Frame] + frames: list[Frame] def __init__(self, vmu_backtrace: DarwinSymbol): backtrace = vmu_backtrace.objc_call('description').py() @@ -313,7 +314,7 @@ def __call__(self, *args, **kwargs): @dataclasses.dataclass class MachPortThreadInfo: - thread_ids: List[int] + thread_ids: list[int] @dataclasses.dataclass @@ -321,7 +322,7 @@ class MachPortInfo: task: int pid: int name: int - rights: List[str] + rights: list[str] ipc_object: int dead: bool proc_name: Optional[str] = None @@ -432,7 +433,7 @@ def loaded_classes(self): yield LoadedClass(name=name, type_name=type_name, binary_path=binary_path) @property - def images(self) -> List[Image]: + def images(self) -> list[Image]: """ get loaded image list """ result = [] @@ -452,11 +453,11 @@ def images(self) -> List[Image]: return result @property - def app_images(self) -> List[Image]: + def app_images(self) -> list[Image]: return [image for image in self.images if APP_SUFFIX in image.path] @property - def threads(self) -> List[Thread]: + def threads(self) -> list[Thread]: result = [] with self._client.safe_malloc(8) as threads: with self._client.safe_malloc(4) as count: @@ -474,7 +475,7 @@ def pid(self) -> int: return self._pid @property - def fds(self) -> List[Fd]: + def fds(self) -> list[Fd]: """ get a list of process opened file descriptors """ result = [] for fdstruct in self.fd_structs: @@ -510,7 +511,7 @@ def fds(self) -> List[Fd]: return result @property - def fd_structs(self) -> List[FdStruct]: + def fd_structs(self) -> list[FdStruct]: """ get a list of process opened file descriptors as raw structs """ result = [] size = self._client.symbols.proc_pidinfo(self.pid, PROC_PIDLISTFDS, 0, 0, 0) @@ -582,7 +583,7 @@ def task_all_info(self): return proc_taskallinfo.parse_stream(pti) @property - def backtraces(self) -> List[Backtrace]: + def backtraces(self) -> list[Backtrace]: result = [] backtraces = self._client.symbols.objc_getClass('VMUSampler').objc_call('sampleAllThreadsOfTask:', self.task) for i in range(backtraces.objc_call('count')): @@ -671,11 +672,11 @@ def parent(self) -> 'Process': return Process(self._client, self.ppid) @property - def environ(self) -> List[str]: + def environ(self) -> list[str]: return self.vmu_proc_info.objc_call('envVars').py() @property - def arguments(self) -> List[str]: + def arguments(self) -> list[str]: return self.vmu_proc_info.objc_call('arguments').py() @property @@ -687,7 +688,7 @@ def procargs2(self) -> Container: return procargs2_t.parse(self.raw_procargs2) @property - def regions(self) -> List[Region]: + def regions(self) -> list[Region]: result = [] # remove the '()' wrapping the list: @@ -804,7 +805,7 @@ def dump_app(self, output_dir: str, chunk_size=CHUNK_SIZE) -> None: output_file.flush() output_file.write((image.address + crypt_offset).peek(crypt_size)) - def get_mach_port_cross_ref_info(self) -> List[MachPortCrossRefInfo]: + def get_mach_port_cross_ref_info(self) -> list[MachPortCrossRefInfo]: """ Get all allocated mach ports and cross-refs to get the recv right owner """ result = [] own_ports = [] @@ -894,7 +895,7 @@ def get_by_name(self, name: str) -> Process: return p raise ArgumentError(f'failed to locate process with name: {name}') - def grep(self, name: str) -> List[Process]: + def grep(self, name: str) -> list[Process]: """ get process list by basename filter """ result = [] proc_list = self.list() @@ -903,7 +904,7 @@ def grep(self, name: str) -> List[Process]: result.append(p) return result - def get_processes_by_listening_port(self, port: int) -> List[Process]: + def get_processes_by_listening_port(self, port: int) -> list[Process]: """ get a process object listening on the specified port """ listening_processes = [] for process in self.list(): @@ -920,7 +921,7 @@ def get_processes_by_listening_port(self, port: int) -> List[Process]: listening_processes.append(process) return listening_processes - def lsof(self) -> Mapping[int, List[Fd]]: + def lsof(self) -> dict[int, list[Fd]]: """ get dictionary of pid to its opened fds """ result = {} for process in self.list(): @@ -935,7 +936,7 @@ def lsof(self) -> Mapping[int, List[Fd]]: return result @path_to_str('path') - def fuser(self, path: str) -> List[Process]: + def fuser(self, path: str) -> list[Process]: """get a list of all processes have an open hande to the specified path """ result = [] proc_list = self.list() @@ -954,7 +955,7 @@ def fuser(self, path: str) -> List[Process]: return result - def list(self) -> List[Process]: + def list(self) -> list[Process]: """ list all currently running processes """ n = self._client.symbols.proc_listallpids(0, 0) pid_buf_size = pid_t.sizeof() * n diff --git a/src/rpcclient/rpcclient/darwin/reports.py b/src/rpcclient/rpcclient/darwin/reports.py index 060378a1..aefaa1ae 100644 --- a/src/rpcclient/rpcclient/darwin/reports.py +++ b/src/rpcclient/rpcclient/darwin/reports.py @@ -1,5 +1,4 @@ from pathlib import Path -from typing import List from rpcclient.darwin.crash_reports import CrashReports @@ -11,7 +10,7 @@ def __init__(self, client, crash_reports_dir): self._client = client self.crash_reports = CrashReports(client, crash_reports_dir) - def get_logs(self, prefix='') -> List[Path]: + def get_logs(self, prefix='') -> list[Path]: result = [] sub_paths = ['var/log', 'Library/Logs'] for sub_path in sub_paths: diff --git a/src/rpcclient/rpcclient/darwin/scpreferences.py b/src/rpcclient/rpcclient/darwin/scpreferences.py index 33530d86..b7f4b363 100644 --- a/src/rpcclient/rpcclient/darwin/scpreferences.py +++ b/src/rpcclient/rpcclient/darwin/scpreferences.py @@ -1,4 +1,3 @@ -import typing from collections import UserDict import IPython @@ -35,7 +34,7 @@ def __init__(self, client, preferences_id: str, ref): self._preferences_id = preferences_id @property - def keys(self) -> typing.List[str]: + def keys(self) -> list[str]: """ wrapper for SCPreferencesCopyKeyList """ return self._client.symbols.SCPreferencesCopyKeyList(self._ref).py() @@ -49,18 +48,18 @@ def set(self, key: str, value): self._set(key, value) self._commit() - def set_dict(self, d: typing.Mapping): + def set_dict(self, d: dict): """ set the entire preference dictionary (clear if already exists) and commit the change """ self._clear() self._update_dict(d) self._commit() - def _update_dict(self, d: typing.Mapping): + def _update_dict(self, d: dict): """ update preference dictionary """ for k, v in d.items(): self._set(k, v) - def update_dict(self, d: typing.Mapping): + def update_dict(self, d: dict): """ update preference dictionary and commit """ self._update_dict(d) self._commit() @@ -79,7 +78,7 @@ def get(self, key: str): """ wrapper for SCPreferencesGetValue """ return self._client.symbols.SCPreferencesGetValue(self._ref, self._client.cf(key)).py() - def get_dict(self) -> typing.Mapping: + def get_dict(self) -> dict: """ get a dictionary representation """ result = {} for k in self.keys: @@ -147,7 +146,7 @@ def open(self, preferences_id: str) -> SCPreference: raise RpcClientException(f'SCPreferencesCreate failed for: {preferences_id}') return SCPreference(self._client, preferences_id, ref) - def get_keys(self, preferences_id: str) -> typing.List[str]: + def get_keys(self, preferences_id: str) -> list[str]: """ get all keys from given preferences_id """ with self.open(preferences_id) as o: return o.keys diff --git a/src/rpcclient/rpcclient/darwin/symbol.py b/src/rpcclient/rpcclient/darwin/symbol.py index e239cdd3..107f8537 100644 --- a/src/rpcclient/rpcclient/darwin/symbol.py +++ b/src/rpcclient/rpcclient/darwin/symbol.py @@ -1,4 +1,4 @@ -from typing import List, Optional +from typing import Optional from osstatus.cache import ErrorCode, get_possible_error_codes @@ -48,5 +48,5 @@ def objc_symbol(self): return self._client.objc_symbol(self) @property - def osstatus(self) -> Optional[List[ErrorCode]]: + def osstatus(self) -> Optional[list[ErrorCode]]: return get_possible_error_codes(self) diff --git a/src/rpcclient/rpcclient/darwin/syslog.py b/src/rpcclient/rpcclient/darwin/syslog.py index b3070806..4ac1d882 100644 --- a/src/rpcclient/rpcclient/darwin/syslog.py +++ b/src/rpcclient/rpcclient/darwin/syslog.py @@ -1,7 +1,6 @@ import datetime import logging import re -from typing import List from cached_property import cached_property @@ -91,11 +90,11 @@ def __init__(self, client, subsystem: str): super().__init__(client, obj) @property - def category_strings(self) -> List[OsLogPreferencesCategory]: + def category_strings(self) -> list[OsLogPreferencesCategory]: return self._object.objc_call('categories').py() @property - def categories(self) -> List[OsLogPreferencesCategory]: + def categories(self) -> list[OsLogPreferencesCategory]: result = [] for name in self.category_strings: result.append(self.get_category(name)) @@ -111,11 +110,11 @@ def __init__(self, client): super().__init__(client, obj) @property - def subsystem_strings(self) -> List[str]: + def subsystem_strings(self) -> list[str]: return self._object.objc_call('subsystems').py() @property - def subsystems(self) -> List[OsLogPreferencesSubsystem]: + def subsystems(self) -> list[OsLogPreferencesSubsystem]: result = [] for name in self.subsystem_strings: result.append(self.get_subsystem(name)) diff --git a/src/rpcclient/rpcclient/darwin/xpc.py b/src/rpcclient/rpcclient/darwin/xpc.py index 793581c6..6131e981 100644 --- a/src/rpcclient/rpcclient/darwin/xpc.py +++ b/src/rpcclient/rpcclient/darwin/xpc.py @@ -1,6 +1,5 @@ from datetime import datetime from functools import lru_cache -from typing import List, Mapping from uuid import UUID from rpcclient.darwin.common import CfSerializable @@ -176,11 +175,11 @@ def send_message_raw(self, service_name: str, message_raw: DarwinSymbol) -> Darw conn = self._connect_to_mach_service(service_name) return self._client.symbols.xpc_connection_send_message_with_reply_sync(conn, message_raw) - def force_run_activities(self, activities: List[str]) -> None: + def force_run_activities(self, activities: list[str]) -> None: self.sharedScheduler.objc_call('forceRunActivities:', self._client.cf(activities)) @property - def loaded_activities(self) -> Mapping: + def loaded_activities(self) -> dict: return self._client.preferences.cf.get_dict('com.apple.xpc.activity2', 'root') def set_activity_base_date(self, name: str, date: datetime) -> None: diff --git a/src/rpcclient/rpcclient/fs.py b/src/rpcclient/rpcclient/fs.py index 7ae668cc..c00f945d 100644 --- a/src/rpcclient/rpcclient/fs.py +++ b/src/rpcclient/rpcclient/fs.py @@ -4,8 +4,9 @@ import posixpath import stat import tempfile +from collections.abc import Generator from pathlib import Path, PosixPath -from typing import Generator, List, Union +from typing import Union from parameter_decorators import path_to_str @@ -295,7 +296,7 @@ def _cp_dir(self, source: Path, dest: Path, force: bool): else: dest_file.write_bytes(src_file.read_bytes()) - def _cp(self, sources: List[Path], dest: Path, recursive: bool, force: bool): + def _cp(self, sources: list[Path], dest: Path, recursive: bool, force: bool): dest_exists = dest.exists() is_dest_dir = dest_exists and dest.is_dir() @@ -496,7 +497,7 @@ def remote_path(self, path: str) -> RemotePath: return RemotePath(path, self._client) @path_to_str('local') - def pull(self, remotes: Union[List[Union[str, Path]], Union[str, Path]], local: str, recursive: bool = False, + def pull(self, remotes: Union[list[Union[str, Path]], Union[str, Path]], local: str, recursive: bool = False, force: bool = False): """ pull complete directory tree """ if not isinstance(remotes, list): @@ -505,7 +506,7 @@ def pull(self, remotes: Union[List[Union[str, Path]], Union[str, Path]], local: self._cp([self.remote_path(remote) for remote in remotes_str], Path(str(local)), recursive, force) @path_to_str('remote') - def push(self, locals: Union[List[Union[str, Path]], Union[str, Path]], remote: str, recursive: bool = False, + def push(self, locals: Union[list[Union[str, Path]], Union[str, Path]], remote: str, recursive: bool = False, force: bool = False): """ push complete directory tree """ if not isinstance(locals, list): @@ -555,12 +556,12 @@ def pwd(self) -> str: return buf @path_to_str('path') - def listdir(self, path: str = '.') -> List[str]: + def listdir(self, path: str = '.') -> list[str]: """ get directory listing for a given dirname """ return [e.name for e in self.scandir(path)] @path_to_str('path') - def scandir(self, path: str = '.') -> List[DirEntry]: + def scandir(self, path: str = '.') -> list[DirEntry]: """ get directory listing for a given dirname """ result = [] for entry in self._client.listdir(path)[2:]: @@ -637,8 +638,7 @@ def walk(self, top: str, topdown=True, onerror=None): if dirs: for d in dirs: - for walk_result in self.walk(posixpath.join(top, d), topdown=topdown, onerror=onerror): - yield walk_result + yield from self.walk(posixpath.join(top, d), topdown=topdown, onerror=onerror) if not topdown: yield top, dirs, files diff --git a/src/rpcclient/rpcclient/ios/accessibility.py b/src/rpcclient/rpcclient/ios/accessibility.py index aa642e7d..25f4d042 100644 --- a/src/rpcclient/rpcclient/ios/accessibility.py +++ b/src/rpcclient/rpcclient/ios/accessibility.py @@ -1,7 +1,7 @@ import dataclasses import time from enum import IntEnum, IntFlag -from typing import List, Optional +from typing import Optional from rpcclient.darwin.symbol import DarwinSymbol from rpcclient.exceptions import ElementNotFoundError, FirstElementNotFoundError, LastElementNotFoundError, \ @@ -215,7 +215,7 @@ def traits(self) -> AXTraits: return AXTraits(self.objc_call('traits').c_uint64) @property - def elements(self) -> List['AXElement']: + def elements(self) -> list['AXElement']: """ get all current displayed elements """ result = [] elements = self.objc_call('explorerElements') @@ -284,14 +284,14 @@ def _next_opaque(self, direction=AXDirection.Next): return element - def _next_elements_with_count(self, count: int) -> List['AXElement']: + def _next_elements_with_count(self, count: int) -> list['AXElement']: elements = self.objc_call('nextElementsWithCount:', count) result = [] for i in range(elements.objc_call('count')): result.append(AXElement.create(elements.objc_call('objectAtIndex:', i), self._client)) return result - def _previous_elements_with_count(self, count: int) -> List['AXElement']: + def _previous_elements_with_count(self, count: int) -> list['AXElement']: elements = self.objc_call('previousElementsWithCount:', count) result = [] for i in range(elements.objc_call('count')): @@ -463,7 +463,7 @@ def _get_element_by_label(self, label: str, auto_scroll=True, draw_frame=True, d raise ElementNotFoundError(f'failed to find AXElement by label: "{label}"') - def press_elements_by_labels(self, labels: List[str], auto_scroll=True, draw_frame=True, timeout=5, + def press_elements_by_labels(self, labels: list[str], auto_scroll=True, draw_frame=True, timeout=5, direction: AXDirection = AXDirection.Next, displayed_only=False): """ press a sequence of labels diff --git a/src/rpcclient/rpcclient/ios/client.py b/src/rpcclient/rpcclient/ios/client.py index b02b3b51..5905f65f 100644 --- a/src/rpcclient/rpcclient/ios/client.py +++ b/src/rpcclient/rpcclient/ios/client.py @@ -33,7 +33,7 @@ def __init__(self, sock, sysname: str, arch, create_socket_cb: typing.Callable): self._radio_preferences = self.symbols.objc_getClass('RadiosPreferences').objc_call('new') @property - def roots(self) -> typing.List[str]: + def roots(self) -> list[str]: """ get a list of all accessible darwin roots when used for lookup of files/preferences/... """ return super().roots + ['/var/mobile'] diff --git a/src/rpcclient/rpcclient/ios/lockdown.py b/src/rpcclient/rpcclient/ios/lockdown.py index 2d3c8912..f17bdce7 100644 --- a/src/rpcclient/rpcclient/ios/lockdown.py +++ b/src/rpcclient/rpcclient/ios/lockdown.py @@ -3,7 +3,6 @@ import posixpath import uuid from datetime import datetime, timedelta -from typing import List, Mapping from rpcclient.darwin.scpreferences import SCPreference @@ -22,7 +21,7 @@ def host_id(self) -> str: return self._host_id @property - def record(self) -> Mapping: + def record(self) -> dict: return plistlib.loads(self._client.fs.read_file(posixpath.join(PAIR_RECORD_PATH, f'{self._host_id}.plist'))) @property @@ -55,7 +54,7 @@ def get_host_id(hostname: str = None) -> str: return str(host_id).upper() @property - def pair_records(self) -> List[PairRecord]: + def pair_records(self) -> list[PairRecord]: """ list pair records """ result = [] for filename in self._client.fs.listdir(PAIR_RECORD_PATH): @@ -63,7 +62,7 @@ def pair_records(self) -> List[PairRecord]: return result @property - def pair_dates(self) -> Mapping: + def pair_dates(self) -> dict: result = {} raw = self._client.preferences.cf.get_dict('com.apple.mobile.ldpair', 'mobile', 'kCFPreferencesAnyHost') for host_id, timestmap in raw.items(): @@ -87,7 +86,7 @@ def get_pair_record_by_hostname(self, hostname: str) -> PairRecord: def get_self_pair_record(self) -> PairRecord: return self.get_pair_record_by_host_id(self.get_host_id()) - def add_pair_record(self, pair_record: Mapping, date: datetime, hostname: str = None): + def add_pair_record(self, pair_record: dict, date: datetime, hostname: str = None): pair_record = dict(pair_record) # remove private key from pair record before adding it pair_record.pop('HostPrivateKey') diff --git a/src/rpcclient/rpcclient/ios/wifi.py b/src/rpcclient/rpcclient/ios/wifi.py index b4607413..bf40954b 100644 --- a/src/rpcclient/rpcclient/ios/wifi.py +++ b/src/rpcclient/rpcclient/ios/wifi.py @@ -1,6 +1,6 @@ import ctypes import logging -from typing import List, Mapping +from typing import Optional from rpcclient.allocated import Allocated from rpcclient.darwin.symbol import DarwinSymbol @@ -40,7 +40,7 @@ def __repr__(self): class WifiScannedNetwork: - def __init__(self, client, interface: DarwinSymbol, network: Mapping): + def __init__(self, client, interface: DarwinSymbol, network: dict): self._client = client self._interface = interface self.network = network @@ -90,7 +90,7 @@ def __init__(self, client, interface, device): self._interface = interface self._device = device - def scan(self, options: Mapping = None) -> List[WifiScannedNetwork]: + def scan(self, options: Optional[dict] = None) -> list[WifiScannedNetwork]: """ perform Wi-Fi scan """ if options is None: @@ -132,7 +132,7 @@ def __init__(self, client): self._client.raise_errno_exception('WiFiManagerClientCreate failed') @property - def saved_networks(self) -> List[WifiSavedNetwork]: + def saved_networks(self) -> list[WifiSavedNetwork]: result = [] network_list = self._client.symbols.WiFiManagerClientCopyNetworks(self._wifi_manager).py() @@ -145,7 +145,7 @@ def saved_networks(self) -> List[WifiSavedNetwork]: return result @property - def interfaces(self) -> List[str]: + def interfaces(self) -> list[str]: """ get a list of all available wifi interfaces """ with self._client.safe_malloc(8) as p_interface: if self._client.symbols.Apple80211Open(p_interface): diff --git a/src/rpcclient/rpcclient/lief.py b/src/rpcclient/rpcclient/lief.py index 89536c41..9b882a3a 100644 --- a/src/rpcclient/rpcclient/lief.py +++ b/src/rpcclient/rpcclient/lief.py @@ -1,5 +1,4 @@ from collections import namedtuple -from typing import Mapping import lief from parameter_decorators import path_to_str @@ -19,7 +18,7 @@ def parse(self, path: str): return lief.parse(f.read()) @path_to_str('path') - def get_symbols(self, path: str) -> Mapping[str, Symbol]: + def get_symbols(self, path: str) -> dict[str, Symbol]: result = {} parsed = self.parse(path) for s in parsed.symbols: diff --git a/src/rpcclient/rpcclient/macos/client.py b/src/rpcclient/rpcclient/macos/client.py index d389d926..f5b14b8e 100644 --- a/src/rpcclient/rpcclient/macos/client.py +++ b/src/rpcclient/rpcclient/macos/client.py @@ -14,7 +14,7 @@ def __init__(self, sock, sysname: str, arch, create_socket_cb: typing.Callable): self.apple_script = AppleScript(self) @property - def roots(self) -> typing.List[str]: + def roots(self) -> list[str]: """ get a list of all accessible darwin roots when used for lookup of files/preferences/... """ result = super().roots diff --git a/src/rpcclient/rpcclient/network.py b/src/rpcclient/rpcclient/network.py index dd87c805..560fef82 100644 --- a/src/rpcclient/rpcclient/network.py +++ b/src/rpcclient/rpcclient/network.py @@ -187,7 +187,7 @@ def gethostbyname(self, name: str) -> Hostentry: return Hostentry(name=result.h_name, aliases=aliases, addresses=addresses) @property - def interfaces(self) -> typing.List[Interface]: + def interfaces(self) -> list[Interface]: """ get current interfaces """ results = [] my_ifaddrs = ifaddrs(self._client) diff --git a/src/rpcclient/rpcclient/protosocket.py b/src/rpcclient/rpcclient/protosocket.py index 8b548774..938b4e4d 100644 --- a/src/rpcclient/rpcclient/protosocket.py +++ b/src/rpcclient/rpcclient/protosocket.py @@ -1,7 +1,6 @@ import socket import struct import threading -from typing import Tuple from rpcclient.exceptions import InvalidServerVersionMagicError, ServerDiedError, ServerResponseError from rpcclient.protos.rpc_pb2 import CmdClose, Command, Handshake, Response @@ -42,7 +41,7 @@ def send_recv(self, sub_command): else: return getattr(response, command_type.lower()) - def _receive(self) -> Tuple[int, bytes]: + def _receive(self) -> tuple[int, bytes]: try: size = struct.unpack(' List[CsInsn]: + def disass(self, size=40) -> list[CsInsn]: """ peek disassembled lines of 'size' bytes """ if self._client.arch == ARCH_ARM64: return list(Cs(CS_ARCH_ARM64, CS_MODE_LITTLE_ENDIAN).disasm(self.peek(size), self)) diff --git a/src/rpcclient/rpcclient/xonshrc.py b/src/rpcclient/rpcclient/xonshrc.py index 4389ff29..ed27b8fa 100644 --- a/src/rpcclient/rpcclient/xonshrc.py +++ b/src/rpcclient/rpcclient/xonshrc.py @@ -10,7 +10,7 @@ from argparse import ArgumentParser from datetime import datetime from pathlib import Path -from typing import Callable, List, Union +from typing import Callable, Union from uuid import UUID import plumbum @@ -76,7 +76,7 @@ def path_completer(xsh, action, completer, alias, command): result = [] for f in client.fs.scandir(dirpath): if is_absolute: - completion_option = str((dirpath / f.name)) + completion_option = str(dirpath / f.name) else: completion_option = str((dirpath / f.name).relative_to(pwd)) try: @@ -100,7 +100,7 @@ def dir_completer(xsh, action, completer, alias, command): result = [] for f in client.fs.scandir(dirpath): if is_absolute: - completion_option = str((dirpath / f.name)) + completion_option = str(dirpath / f.name) else: completion_option = str((dirpath / f.name).relative_to(pwd)) try: @@ -343,7 +343,7 @@ def _rpc_list_elements(self): f'{{RESET}}', file=sys.stdout) - def _rpc_press_elements(self, label: Annotated[List[str], Arg(nargs='+', completer=element_completer)]): + def _rpc_press_elements(self, label: Annotated[list[str], Arg(nargs='+', completer=element_completer)]): """ press labels list by given order """ @@ -362,7 +362,7 @@ def _rpc_disable_accessibility(self): self.client.accessibility.enabled = False def _rpc_press_keys(self, key: Annotated[ - List[str], Arg(nargs='+', + list[str], Arg(nargs='+', completer=lambda xsh, action, completer, alias, command: ['power', 'home', 'volup', 'voldown', 'mute'])]): """ @@ -481,7 +481,7 @@ def _rpc_cd(self, path: Annotated[str, Arg(completer=dir_completer)]): """ self.client.fs.chdir(path) - def _rpc_rm(self, path: Annotated[List[str], Arg(nargs='+', completer=path_completer)], recursive=False, + def _rpc_rm(self, path: Annotated[list[str], Arg(nargs='+', completer=path_completer)], recursive=False, force=False): """ remove files @@ -522,7 +522,7 @@ def _rpc_mkdir(self, filename: Annotated[str, Arg(completer=path_completer)], pa """ self.client.fs.mkdir(filename, mode=int(mode, 8), parents=parents) - def _rpc_cat(self, filename: Annotated[List[str], Arg(nargs='+', completer=path_completer)]): + def _rpc_cat(self, filename: Annotated[list[str], Arg(nargs='+', completer=path_completer)]): """ read a list of files """ @@ -536,7 +536,7 @@ def _rpc_cat(self, filename: Annotated[List[str], Arg(nargs='+', completer=path_ except UnicodeDecodeError: return str(buf) - def _rpc_bat(self, filename: Annotated[List[str], Arg(completer=path_completer)]): + def _rpc_bat(self, filename: Annotated[list[str], Arg(completer=path_completer)]): """ "bat" (improved-cat) given file """ @@ -544,7 +544,7 @@ def _rpc_bat(self, filename: Annotated[List[str], Arg(completer=path_completer)] os.system(f'bat "{f}"') def _rpc_pull( - self, files: Annotated[List[str], + self, files: Annotated[list[str], Arg(nargs='+', completer=path_completer)], recursive: bool = False, force: bool = False): """ @@ -563,7 +563,7 @@ def _rpc_pull( return self._pull(files, local, recursive, force) def _rpc_push( - self, files: Annotated[List[str], + self, files: Annotated[list[str], Arg(nargs='+', completer=path_completer)], recursive: bool = False, force: bool = False): """ @@ -710,7 +710,7 @@ def _edit_remotely(self, remote): def _relative_path(self, filename): return posixpath.join(self._rpc_pwd(), filename) - def _listdir(self, path: str) -> List[str]: + def _listdir(self, path: str) -> list[str]: return self.client.fs.listdir(path) def _pull(self, remote_filename, local_filename, recursive: bool = False, force: bool = False): diff --git a/src/rpcclient/tests/test_objc_symbol.py b/src/rpcclient/tests/test_objc_symbol.py index 20c9f6e1..1bfbfba7 100644 --- a/src/rpcclient/tests/test_objc_symbol.py +++ b/src/rpcclient/tests/test_objc_symbol.py @@ -11,16 +11,16 @@ def test_method_by_method_name(client): str1 = NSString.stringWithCString_encoding_('Taylor Swift', ascii_encoding).objc_symbol assert str1.cStringUsingEncoding_(ascii_encoding).peek_str() == 'Taylor Swift' assert str1.length == len('Taylor Swift') - assert str1.lowercaseString().cStringUsingEncoding_(ascii_encoding).peek_str() == 'taylor swift' - assert str1.uppercaseString().cStringUsingEncoding_(ascii_encoding).peek_str() == 'TAYLOR SWIFT' + assert str1.lowercaseString().objc_symbol.cStringUsingEncoding_(ascii_encoding).peek_str() == 'taylor swift' + assert str1.uppercaseString().objc_symbol.cStringUsingEncoding_(ascii_encoding).peek_str() == 'TAYLOR SWIFT' def test_calling_property(client): - d = client.symbols.objc_getClass('NSMutableDictionary').objc_call('new').objc_symbol + d = client.symbols.objc_getClass('NSMutableDictionary').objc_call('new') # call method - d.setObject_forKey_(client.cf('value'), client.cf('key')) + d.objc_call('setObject:forKey:', client.cf('value'), client.cf('key')) # call property - assert '{\n key = value;\n}' == d.description.py() + assert '{\n key = value;\n}' == d.objc_symbol.description.py() def test_set_implementation(client):