Skip to content

System & device services

Services for device state, activation, notifications, power, and recovery.

pymobiledevice3.services.amfi.AmfiService

Control Developer Mode on the device through the AMFI (Apple Mobile File Integrity) lockdown service.

Each method opens its own connection to the com.apple.amfi.lockdown service and sends a single action request. The actions correspond to revealing the Developer Mode toggle in Settings, requesting that it be enabled, and confirming the post-restart prompt.

Source code in pymobiledevice3/services/amfi.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
class AmfiService:
    """
    Control Developer Mode on the device through the AMFI (Apple Mobile File Integrity) lockdown service.

    Each method opens its own connection to the ``com.apple.amfi.lockdown`` service and sends a single
    ``action`` request. The actions correspond to revealing the Developer Mode toggle in Settings,
    requesting that it be enabled, and confirming the post-restart prompt.
    """

    DEVELOPER_MODE_REVEAL = 0
    DEVELOPER_MODE_ENABLE = 1
    DEVELOPER_MODE_ACCEPT = 2

    SERVICE_NAME = "com.apple.amfi.lockdown"

    def __init__(self, lockdown: LockdownServiceProvider) -> None:
        self._lockdown = lockdown
        self._logger = logging.getLogger(self.__module__)

    async def reveal_developer_mode_option_in_ui(self):
        """
        Make the Developer Mode toggle visible in the device's Settings UI.

        Sends the reveal action, which causes AMFI to create an empty file at its
        ``AMFIShowOverridePath``.

        :raises PyMobileDevice3Exception: if the device does not report success.
        """
        service = await self._lockdown.start_lockdown_service(self.SERVICE_NAME)
        resp = await service.send_recv_plist({"action": self.DEVELOPER_MODE_REVEAL})
        if not resp.get("success"):
            raise PyMobileDevice3Exception(f"create_AMFIShowOverridePath() failed with: {resp}")

    async def enable_developer_mode(self, enable_post_restart=True):
        """
        Request that Developer Mode be enabled on the device.

        Sends the enable action. The device then reboots to apply the change. When
        ``enable_post_restart`` is True, this method keeps the connection alive via a heartbeat,
        waits for the device to disconnect and reconnect over usbmux, and then answers the
        post-restart confirmation prompt by calling `enable_developer_mode_post_restart`.

        :param enable_post_restart: when True, wait for the device to restart and automatically
            confirm the final prompt; when False, return immediately after the enable request.
        :raises DeviceHasPasscodeSetError: if the device has a passcode set, which blocks the operation.
        :raises AmfiError: if AMFI returns any other error.
        :raises DeveloperModeError: if the enable or post-restart confirmation request does not succeed.
        """
        service = await self._lockdown.start_lockdown_service(self.SERVICE_NAME)
        resp = await service.send_recv_plist({"action": self.DEVELOPER_MODE_ENABLE})
        error = resp.get("Error")

        if error is not None:
            if error == "Device has a passcode set":
                raise DeviceHasPasscodeSetError()
            raise AmfiError(error)

        if not resp.get("success"):
            raise DeveloperModeError(f"enable_developer_mode(): {resp}")

        if not enable_post_restart:
            return

        try:
            await HeartbeatService(self._lockdown).start()
        except (ConnectionTerminatedError, BrokenPipeError, IncompleteReadError):
            self._logger.debug("device disconnected, awaiting reconnect")

        self._lockdown = await retry_create_using_usbmux(None, serial=self._lockdown.udid)
        await self.enable_developer_mode_post_restart()

    async def enable_developer_mode_post_restart(self):
        """
        Confirm the Developer Mode prompt shown after the device restarts.

        Sends the accept action, answering the post-restart confirmation prompt affirmatively.

        :raises DeveloperModeError: if the device does not report success.
        """
        service = await self._lockdown.start_lockdown_service(self.SERVICE_NAME)
        resp = await service.send_recv_plist({"action": self.DEVELOPER_MODE_ACCEPT})
        if not resp.get("success"):
            raise DeveloperModeError(f"enable_developer_mode_post_restart() failed: {resp}")

reveal_developer_mode_option_in_ui async 

reveal_developer_mode_option_in_ui()

Make the Developer Mode toggle visible in the device's Settings UI.

Sends the reveal action, which causes AMFI to create an empty file at its AMFIShowOverridePath.

Raises:

Type Description
PyMobileDevice3Exception

if the device does not report success.
Source code in pymobiledevice3/services/amfi.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
async def reveal_developer_mode_option_in_ui(self):
    """
    Make the Developer Mode toggle visible in the device's Settings UI.

    Sends the reveal action, which causes AMFI to create an empty file at its
    ``AMFIShowOverridePath``.

    :raises PyMobileDevice3Exception: if the device does not report success.
    """
    service = await self._lockdown.start_lockdown_service(self.SERVICE_NAME)
    resp = await service.send_recv_plist({"action": self.DEVELOPER_MODE_REVEAL})
    if not resp.get("success"):
        raise PyMobileDevice3Exception(f"create_AMFIShowOverridePath() failed with: {resp}")

enable_developer_mode async 

enable_developer_mode(enable_post_restart=True)

Request that Developer Mode be enabled on the device.

Sends the enable action. The device then reboots to apply the change. When enable_post_restart is True, this method keeps the connection alive via a heartbeat, waits for the device to disconnect and reconnect over usbmux, and then answers the post-restart confirmation prompt by calling enable_developer_mode_post_restart.

Parameters:

Name Type Description Default
enable_post_restart

when True, wait for the device to restart and automatically confirm the final prompt; when False, return immediately after the enable request.

 True

Raises:

Type Description
DeviceHasPasscodeSetError

if the device has a passcode set, which blocks the operation.
AmfiError

if AMFI returns any other error.
DeveloperModeError

if the enable or post-restart confirmation request does not succeed.
Source code in pymobiledevice3/services/amfi.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
async def enable_developer_mode(self, enable_post_restart=True):
    """
    Request that Developer Mode be enabled on the device.

    Sends the enable action. The device then reboots to apply the change. When
    ``enable_post_restart`` is True, this method keeps the connection alive via a heartbeat,
    waits for the device to disconnect and reconnect over usbmux, and then answers the
    post-restart confirmation prompt by calling `enable_developer_mode_post_restart`.

    :param enable_post_restart: when True, wait for the device to restart and automatically
        confirm the final prompt; when False, return immediately after the enable request.
    :raises DeviceHasPasscodeSetError: if the device has a passcode set, which blocks the operation.
    :raises AmfiError: if AMFI returns any other error.
    :raises DeveloperModeError: if the enable or post-restart confirmation request does not succeed.
    """
    service = await self._lockdown.start_lockdown_service(self.SERVICE_NAME)
    resp = await service.send_recv_plist({"action": self.DEVELOPER_MODE_ENABLE})
    error = resp.get("Error")

    if error is not None:
        if error == "Device has a passcode set":
            raise DeviceHasPasscodeSetError()
        raise AmfiError(error)

    if not resp.get("success"):
        raise DeveloperModeError(f"enable_developer_mode(): {resp}")

    if not enable_post_restart:
        return

    try:
        await HeartbeatService(self._lockdown).start()
    except (ConnectionTerminatedError, BrokenPipeError, IncompleteReadError):
        self._logger.debug("device disconnected, awaiting reconnect")

    self._lockdown = await retry_create_using_usbmux(None, serial=self._lockdown.udid)
    await self.enable_developer_mode_post_restart()

enable_developer_mode_post_restart async 

enable_developer_mode_post_restart()

Confirm the Developer Mode prompt shown after the device restarts.

Sends the accept action, answering the post-restart confirmation prompt affirmatively.

Raises:

Type Description
DeveloperModeError

if the device does not report success.
Source code in pymobiledevice3/services/amfi.py 
88
89
90
91
92
93
94
95
96
97
98
99
async def enable_developer_mode_post_restart(self):
    """
    Confirm the Developer Mode prompt shown after the device restarts.

    Sends the accept action, answering the post-restart confirmation prompt affirmatively.

    :raises DeveloperModeError: if the device does not report success.
    """
    service = await self._lockdown.start_lockdown_service(self.SERVICE_NAME)
    resp = await service.send_recv_plist({"action": self.DEVELOPER_MODE_ACCEPT})
    if not resp.get("success"):
        raise DeveloperModeError(f"enable_developer_mode_post_restart() failed: {resp}")

pymobiledevice3.services.mobile_activation.MobileActivationService

Drive iOS device activation against Apple's activation servers via com.apple.mobileactivationd.

Orchestrates the full activation handshake: it gathers activation information from the device, exchanges it with Apple's activation and DRM-handshake endpoints over HTTP, optionally prompts the user for Apple ID credentials when the server requests them, and writes the resulting activation record back to the device. Both the modern session-based flow and the legacy lockdown flow are supported, with the session flow preferred when available.

A fresh lockdown service connection is opened for each request rather than reusing one, so this class does not inherit from the shared service base.

Source code in pymobiledevice3/services/mobile_activation.py 
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
class MobileActivationService:
    """
    Drive iOS device activation against Apple's activation servers via ``com.apple.mobileactivationd``.

    Orchestrates the full activation handshake: it gathers activation information from the device,
    exchanges it with Apple's activation and DRM-handshake endpoints over HTTP, optionally prompts
    the user for Apple ID credentials when the server requests them, and writes the resulting
    activation record back to the device. Both the modern session-based flow and the legacy
    lockdown flow are supported, with the session flow preferred when available.

    A fresh lockdown service connection is opened for each request rather than reusing one, so this
    class does not inherit from the shared service base.
    """

    SERVICE_NAME = "com.apple.mobileactivationd"

    def __init__(self, lockdown: LockdownServiceProvider) -> None:
        self.lockdown = lockdown
        self.logger = logging.getLogger(__name__)

    async def state(self):
        """
        Return the device's current activation state.

        Queries the activation daemon for the activation state, falling back to the lockdown
        ``ActivationState`` value if the daemon request fails.

        :returns: the activation state (e.g. ``"Unactivated"`` or ``"Activated"``).
        """
        try:
            return (await self.send_command("GetActivationStateRequest"))["Value"]
        except Exception:
            return await self.lockdown.get_value(key="ActivationState")

    async def wait_for_activation_session(self):
        """
        Block until the device produces a fresh activation session handshake.

        Repeatedly requests activation session info until the handshake request message changes,
        indicating the device is ready with a new session. Returns immediately if the device does
        not support the session-based flow.
        """
        try:
            blob = await self.create_activation_session_info()
        except Exception:
            return
        handshake_request_message = blob["HandshakeRequestMessage"]
        while handshake_request_message == blob["HandshakeRequestMessage"]:
            blob = await self.create_activation_session_info()

    @staticmethod
    def _get_activation_form_from_response(content: str) -> ActivationForm:
        root = ET.fromstring(content)
        title = root.find("page/navigationBar").get("title")
        description = root.find("page/tableView/section/footer").text
        fields = []
        for editable in root.findall("page//editableTextRow"):
            fields.append(
                Field(
                    id=editable.get("id"),
                    label=editable.get("label"),
                    placeholder=editable.get("placeholder"),
                    secure=bool(editable.get("secure", False)),
                )
            )
        server_info = {}
        for k, v in root.find("serverInfo").items():
            server_info[k] = v
        return ActivationForm(title=title, description=description, fields=fields, server_info=server_info)

    async def activate(self, skip_apple_id_query: bool = False) -> None:
        """
        Activate the device end-to-end against Apple's activation servers.

        Does nothing if the device is already activated. Otherwise it collects activation info from
        the device (preferring the session-based flow, including a DRM handshake with Apple when
        available), posts it to the activation server, and applies the returned activation record.
        If the server responds with a BuddyML form requesting Apple ID credentials, the user is
        prompted interactively and the form is resubmitted. On success, the device's
        ``ActivationStateAcknowledged`` value is set.

        :param skip_apple_id_query: when True, do not prompt for Apple ID credentials; instead treat
            a credentials request as an iCloud lock and raise.
        :raises MobileActivationException: if the device is iCloud locked (with
            ``skip_apple_id_query`` set) or the activation server response is invalid.
        """
        if await self.state() != "Unactivated":
            self.logger.error("Device is already activated!")
            return

        try:
            blob = await self.create_activation_session_info()
            session_mode = True
        except Exception:
            session_mode = False

        # create drmHandshake request with blob from device
        headers = {"Content-Type": "application/x-apple-plist"}
        headers.update(DEFAULT_HEADERS)
        if session_mode:
            content, headers = self.post(
                ACTIVATION_DRM_HANDSHAKE_DEFAULT_URL, data=plistlib.dumps(blob), headers=headers
            )

        activation_request = {}
        if session_mode:
            activation_info = await self.create_activation_info_with_session(content)
        else:
            activation_info = await self.lockdown.get_value(key="ActivationInfo")
            activation_request.update({
                "InStoreActivation": False,
                "AppleSerialNumber": await self.lockdown.get_value(key="SerialNumber"),
            })
            if self.lockdown.all_values.get("TelephonyCapability"):
                req_pair = {
                    "IMEI": "InternationalMobileEquipmentIdentity",
                    "MEID": "MobileEquipmentIdentifier",
                    "IMSI": "InternationalMobileSubscriberIdentity",
                    "ICCID": "IntegratedCircuitCardIdentity",
                }

                has_meid = False
                for k, v in req_pair.items():
                    lv = self.lockdown.all_values.get(v)
                    if lv is not None:
                        activation_request.update({k: lv})
                        continue
                    else:
                        self.logger.warn(f"Unable to get {k} from lockdownd")
                        if k == "MEID" and has_meid:
                            # Something is wrong if both IMEI & MEID is missing
                            raise MobileActivationException("Unable to obtain both IMEI and MEID")

                    # Either IMEI or MEID, or both
                    if k == "IMEI":
                        has_meid = lv is None
        activation_request.update({"activation-info": plistlib.dumps(activation_info)})

        content, headers = self.post(ACTIVATION_DEFAULT_URL, data=activation_request)
        content_type = headers["Content-Type"]

        if content_type == "application/x-buddyml":
            if skip_apple_id_query:
                raise MobileActivationException("Device is iCloud locked")
            try:
                activation_form = self._get_activation_form_from_response(content.decode())
            except Exception as e:
                raise MobileActivationException("Activation server response is invalid") from e
            else:
                click.secho(activation_form.title, bold=True)
                click.secho(activation_form.description)
                fields = []
                for field in activation_form.fields:
                    if field.secure:
                        fields.append(inquirer3.Password(name=field.id, message=f"{field.label}"))
                    else:
                        fields.append(inquirer3.Text(name=field.id, message=f"{field.label}"))
                data = inquirer3.prompt(fields)
                data.update(activation_form.server_info)
                content, headers = self.post(ACTIVATION_DEFAULT_URL, data=data)
                content_type = headers["Content-Type"]

        assert content_type == "text/xml"
        if session_mode:
            await self.activate_with_session(content, headers)
        else:
            await self.activate_with_lockdown(content)

        # set ActivationStateAcknowledged if we succeeded
        await self.lockdown.set_value(True, key="ActivationStateAcknowledged")

    async def deactivate(self):
        """
        Deactivate the device.

        Sends a ``DeactivateRequest`` to the activation daemon, falling back to the lockdown
        ``Deactivate`` request if that fails.

        :returns: the response from whichever deactivation path succeeds.
        """
        try:
            return await self.send_command("DeactivateRequest")
        except Exception:
            return await self.lockdown._request_async("Deactivate")

    async def create_activation_session_info(self):
        """
        Request the device's session-based activation handshake info.

        Sends a ``CreateTunnel1SessionInfoRequest`` to the activation daemon.

        :returns: the session info blob (the ``Value`` entry), including the handshake request message.
        :raises MobileActivationException: if the daemon returns an error.
        """
        response = await self.send_command("CreateTunnel1SessionInfoRequest")
        error = response.get("Error")
        if error is not None:
            raise MobileActivationException(f"Mobile activation can not be done due to: {response}")
        return response["Value"]

    async def create_activation_info_with_session(self, handshake_response):
        """
        Build the device's activation info from a completed DRM handshake.

        Sends a ``CreateTunnel1ActivationInfoRequest`` to the activation daemon, passing the DRM
        handshake response received from Apple.

        :param handshake_response: the DRM handshake response content returned by Apple's server.
        :returns: the activation info blob (the ``Value`` entry) to post to the activation server.
        :raises MobileActivationException: if the daemon returns an error.
        """
        response = await self.send_command("CreateTunnel1ActivationInfoRequest", handshake_response)
        error = response.get("Error")
        if error is not None:
            raise MobileActivationException(f"Mobile activation can not be done due to: {response}")
        return response["Value"]

    async def activate_with_lockdown(self, activation_record):
        """
        Apply an activation record to the device using the legacy lockdown flow.

        Parses the activation record, extracts the ``iphone-activation`` or ``device-activation``
        node, and submits its activation record via the lockdown ``Activate`` request.

        :param activation_record: the serialized activation record returned by the activation server.
        :raises MobileActivationException: if the activation record does not contain a recognized
            activation node.
        """
        record = plistlib.loads(activation_record)
        node = record.get("iphone-activation")
        if node is None:
            node = record.get("device-activation")
        if node is None:
            raise MobileActivationException("Activation record received is invalid")

        await self.lockdown._request_async("Activate", {"ActivationRecord": node.get("activation-record")})

    async def activate_with_session(self, activation_record, headers):
        """
        Apply an activation record to the device using the session-based flow.

        Sends a ``HandleActivationInfoWithSessionRequest`` to the activation daemon, including the
        activation record and, if provided, the HTTP response headers from the activation server.

        :param activation_record: the activation record returned by the activation server.
        :param headers: HTTP response headers from the activation server, forwarded as
            ``ActivationResponseHeaders``; may be falsy to omit them.
        :returns: the daemon's response to the request.
        """
        data = {
            "Command": "HandleActivationInfoWithSessionRequest",
            "Value": activation_record,
        }
        if headers:
            data["ActivationResponseHeaders"] = dict(headers)
        async with _aclosing(await self.lockdown.start_lockdown_service(self.SERVICE_NAME)) as service:
            return await service.send_recv_plist(data)

    async def send_command(self, command: str, value: str = ""):
        """
        Send a single command to the activation daemon over a fresh service connection.

        Opens the ``com.apple.mobileactivationd`` service, sends the command (with an optional
        ``Value``), and returns the response. The connection is closed afterwards.

        :param command: the activation daemon command name.
        :param value: optional value sent alongside the command; omitted when empty.
        :returns: the daemon's response plist.
        """
        data = {"Command": command}
        if value:
            data["Value"] = value
        async with _aclosing(await self.lockdown.start_lockdown_service(self.SERVICE_NAME)) as service:
            return await service.send_recv_plist(data)

    def post(
        self, url: str, data: dict, headers: Optional[CaseInsensitiveDict[str, str]] = None
    ) -> tuple[bytes, CaseInsensitiveDict[str]]:
        """
        Perform an HTTP POST to an activation server endpoint.

        :param url: the activation endpoint to post to.
        :param data: the form data or request body to send.
        :param headers: optional request headers; the module's default activation headers are used
            when omitted.
        :returns: a tuple of the response body bytes and the response headers.
        """
        if headers is None:
            headers = DEFAULT_HEADERS

        resp = requests.post(url, data=data, headers=headers)
        return resp.content, resp.headers

state async 

state()

Return the device's current activation state.

Queries the activation daemon for the activation state, falling back to the lockdown ActivationState value if the daemon request fails.

Returns:

Type Description

the activation state (e.g. "Unactivated" or "Activated").
Source code in pymobiledevice3/services/mobile_activation.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
async def state(self):
    """
    Return the device's current activation state.

    Queries the activation daemon for the activation state, falling back to the lockdown
    ``ActivationState`` value if the daemon request fails.

    :returns: the activation state (e.g. ``"Unactivated"`` or ``"Activated"``).
    """
    try:
        return (await self.send_command("GetActivationStateRequest"))["Value"]
    except Exception:
        return await self.lockdown.get_value(key="ActivationState")

wait_for_activation_session async 

wait_for_activation_session()

Block until the device produces a fresh activation session handshake.

Repeatedly requests activation session info until the handshake request message changes, indicating the device is ready with a new session. Returns immediately if the device does not support the session-based flow.

Source code in pymobiledevice3/services/mobile_activation.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
async def wait_for_activation_session(self):
    """
    Block until the device produces a fresh activation session handshake.

    Repeatedly requests activation session info until the handshake request message changes,
    indicating the device is ready with a new session. Returns immediately if the device does
    not support the session-based flow.
    """
    try:
        blob = await self.create_activation_session_info()
    except Exception:
        return
    handshake_request_message = blob["HandshakeRequestMessage"]
    while handshake_request_message == blob["HandshakeRequestMessage"]:
        blob = await self.create_activation_session_info()

activate async 

activate(skip_apple_id_query: bool = False) -> None

Activate the device end-to-end against Apple's activation servers.

Does nothing if the device is already activated. Otherwise it collects activation info from the device (preferring the session-based flow, including a DRM handshake with Apple when available), posts it to the activation server, and applies the returned activation record. If the server responds with a BuddyML form requesting Apple ID credentials, the user is prompted interactively and the form is resubmitted. On success, the device's ActivationStateAcknowledged value is set.

Parameters:

Name Type Description Default
skip_apple_id_query bool

when True, do not prompt for Apple ID credentials; instead treat a credentials request as an iCloud lock and raise.

 False

Raises:

Type Description
MobileActivationException

if the device is iCloud locked (with skip_apple_id_query set) or the activation server response is invalid.
Source code in pymobiledevice3/services/mobile_activation.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
async def activate(self, skip_apple_id_query: bool = False) -> None:
    """
    Activate the device end-to-end against Apple's activation servers.

    Does nothing if the device is already activated. Otherwise it collects activation info from
    the device (preferring the session-based flow, including a DRM handshake with Apple when
    available), posts it to the activation server, and applies the returned activation record.
    If the server responds with a BuddyML form requesting Apple ID credentials, the user is
    prompted interactively and the form is resubmitted. On success, the device's
    ``ActivationStateAcknowledged`` value is set.

    :param skip_apple_id_query: when True, do not prompt for Apple ID credentials; instead treat
        a credentials request as an iCloud lock and raise.
    :raises MobileActivationException: if the device is iCloud locked (with
        ``skip_apple_id_query`` set) or the activation server response is invalid.
    """
    if await self.state() != "Unactivated":
        self.logger.error("Device is already activated!")
        return

    try:
        blob = await self.create_activation_session_info()
        session_mode = True
    except Exception:
        session_mode = False

    # create drmHandshake request with blob from device
    headers = {"Content-Type": "application/x-apple-plist"}
    headers.update(DEFAULT_HEADERS)
    if session_mode:
        content, headers = self.post(
            ACTIVATION_DRM_HANDSHAKE_DEFAULT_URL, data=plistlib.dumps(blob), headers=headers
        )

    activation_request = {}
    if session_mode:
        activation_info = await self.create_activation_info_with_session(content)
    else:
        activation_info = await self.lockdown.get_value(key="ActivationInfo")
        activation_request.update({
            "InStoreActivation": False,
            "AppleSerialNumber": await self.lockdown.get_value(key="SerialNumber"),
        })
        if self.lockdown.all_values.get("TelephonyCapability"):
            req_pair = {
                "IMEI": "InternationalMobileEquipmentIdentity",
                "MEID": "MobileEquipmentIdentifier",
                "IMSI": "InternationalMobileSubscriberIdentity",
                "ICCID": "IntegratedCircuitCardIdentity",
            }

            has_meid = False
            for k, v in req_pair.items():
                lv = self.lockdown.all_values.get(v)
                if lv is not None:
                    activation_request.update({k: lv})
                    continue
                else:
                    self.logger.warn(f"Unable to get {k} from lockdownd")
                    if k == "MEID" and has_meid:
                        # Something is wrong if both IMEI & MEID is missing
                        raise MobileActivationException("Unable to obtain both IMEI and MEID")

                # Either IMEI or MEID, or both
                if k == "IMEI":
                    has_meid = lv is None
    activation_request.update({"activation-info": plistlib.dumps(activation_info)})

    content, headers = self.post(ACTIVATION_DEFAULT_URL, data=activation_request)
    content_type = headers["Content-Type"]

    if content_type == "application/x-buddyml":
        if skip_apple_id_query:
            raise MobileActivationException("Device is iCloud locked")
        try:
            activation_form = self._get_activation_form_from_response(content.decode())
        except Exception as e:
            raise MobileActivationException("Activation server response is invalid") from e
        else:
            click.secho(activation_form.title, bold=True)
            click.secho(activation_form.description)
            fields = []
            for field in activation_form.fields:
                if field.secure:
                    fields.append(inquirer3.Password(name=field.id, message=f"{field.label}"))
                else:
                    fields.append(inquirer3.Text(name=field.id, message=f"{field.label}"))
            data = inquirer3.prompt(fields)
            data.update(activation_form.server_info)
            content, headers = self.post(ACTIVATION_DEFAULT_URL, data=data)
            content_type = headers["Content-Type"]

    assert content_type == "text/xml"
    if session_mode:
        await self.activate_with_session(content, headers)
    else:
        await self.activate_with_lockdown(content)

    # set ActivationStateAcknowledged if we succeeded
    await self.lockdown.set_value(True, key="ActivationStateAcknowledged")

deactivate async 

deactivate()

Deactivate the device.

Sends a DeactivateRequest to the activation daemon, falling back to the lockdown Deactivate request if that fails.

Returns:

Type Description

the response from whichever deactivation path succeeds.
Source code in pymobiledevice3/services/mobile_activation.py 
226
227
228
229
230
231
232
233
234
235
236
237
238
async def deactivate(self):
    """
    Deactivate the device.

    Sends a ``DeactivateRequest`` to the activation daemon, falling back to the lockdown
    ``Deactivate`` request if that fails.

    :returns: the response from whichever deactivation path succeeds.
    """
    try:
        return await self.send_command("DeactivateRequest")
    except Exception:
        return await self.lockdown._request_async("Deactivate")

create_activation_session_info async 

create_activation_session_info()

Request the device's session-based activation handshake info.

Sends a CreateTunnel1SessionInfoRequest to the activation daemon.

Returns:

Type Description

the session info blob (the Value entry), including the handshake request message.

Raises:

Type Description
MobileActivationException

if the daemon returns an error.
Source code in pymobiledevice3/services/mobile_activation.py 
240
241
242
243
244
245
246
247
248
249
250
251
252
253
async def create_activation_session_info(self):
    """
    Request the device's session-based activation handshake info.

    Sends a ``CreateTunnel1SessionInfoRequest`` to the activation daemon.

    :returns: the session info blob (the ``Value`` entry), including the handshake request message.
    :raises MobileActivationException: if the daemon returns an error.
    """
    response = await self.send_command("CreateTunnel1SessionInfoRequest")
    error = response.get("Error")
    if error is not None:
        raise MobileActivationException(f"Mobile activation can not be done due to: {response}")
    return response["Value"]

create_activation_info_with_session async 

create_activation_info_with_session(handshake_response)

Build the device's activation info from a completed DRM handshake.

Sends a CreateTunnel1ActivationInfoRequest to the activation daemon, passing the DRM handshake response received from Apple.

Parameters:

Name Type Description Default
handshake_response

the DRM handshake response content returned by Apple's server.

 required

Returns:

Type Description

the activation info blob (the Value entry) to post to the activation server.

Raises:

Type Description
MobileActivationException

if the daemon returns an error.
Source code in pymobiledevice3/services/mobile_activation.py 
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
async def create_activation_info_with_session(self, handshake_response):
    """
    Build the device's activation info from a completed DRM handshake.

    Sends a ``CreateTunnel1ActivationInfoRequest`` to the activation daemon, passing the DRM
    handshake response received from Apple.

    :param handshake_response: the DRM handshake response content returned by Apple's server.
    :returns: the activation info blob (the ``Value`` entry) to post to the activation server.
    :raises MobileActivationException: if the daemon returns an error.
    """
    response = await self.send_command("CreateTunnel1ActivationInfoRequest", handshake_response)
    error = response.get("Error")
    if error is not None:
        raise MobileActivationException(f"Mobile activation can not be done due to: {response}")
    return response["Value"]

activate_with_lockdown async 

activate_with_lockdown(activation_record)

Apply an activation record to the device using the legacy lockdown flow.

Parses the activation record, extracts the iphone-activation or device-activation node, and submits its activation record via the lockdown Activate request.

Parameters:

Name Type Description Default
activation_record

the serialized activation record returned by the activation server.

 required

Raises:

Type Description
MobileActivationException

if the activation record does not contain a recognized activation node.
Source code in pymobiledevice3/services/mobile_activation.py 
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
async def activate_with_lockdown(self, activation_record):
    """
    Apply an activation record to the device using the legacy lockdown flow.

    Parses the activation record, extracts the ``iphone-activation`` or ``device-activation``
    node, and submits its activation record via the lockdown ``Activate`` request.

    :param activation_record: the serialized activation record returned by the activation server.
    :raises MobileActivationException: if the activation record does not contain a recognized
        activation node.
    """
    record = plistlib.loads(activation_record)
    node = record.get("iphone-activation")
    if node is None:
        node = record.get("device-activation")
    if node is None:
        raise MobileActivationException("Activation record received is invalid")

    await self.lockdown._request_async("Activate", {"ActivationRecord": node.get("activation-record")})

activate_with_session async 

activate_with_session(activation_record, headers)

Apply an activation record to the device using the session-based flow.

Sends a HandleActivationInfoWithSessionRequest to the activation daemon, including the activation record and, if provided, the HTTP response headers from the activation server.

Parameters:

Name Type Description Default
activation_record

the activation record returned by the activation server.

 required
headers

HTTP response headers from the activation server, forwarded as ActivationResponseHeaders; may be falsy to omit them.

 required

Returns:

Type Description

the daemon's response to the request.
Source code in pymobiledevice3/services/mobile_activation.py 
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
async def activate_with_session(self, activation_record, headers):
    """
    Apply an activation record to the device using the session-based flow.

    Sends a ``HandleActivationInfoWithSessionRequest`` to the activation daemon, including the
    activation record and, if provided, the HTTP response headers from the activation server.

    :param activation_record: the activation record returned by the activation server.
    :param headers: HTTP response headers from the activation server, forwarded as
        ``ActivationResponseHeaders``; may be falsy to omit them.
    :returns: the daemon's response to the request.
    """
    data = {
        "Command": "HandleActivationInfoWithSessionRequest",
        "Value": activation_record,
    }
    if headers:
        data["ActivationResponseHeaders"] = dict(headers)
    async with _aclosing(await self.lockdown.start_lockdown_service(self.SERVICE_NAME)) as service:
        return await service.send_recv_plist(data)

send_command async 

send_command(command: str, value: str = '')

Send a single command to the activation daemon over a fresh service connection.

Opens the com.apple.mobileactivationd service, sends the command (with an optional Value), and returns the response. The connection is closed afterwards.

Parameters:

Name Type Description Default
command str

the activation daemon command name.

 required
value str

optional value sent alongside the command; omitted when empty.

 ''

Returns:

Type Description

the daemon's response plist.
Source code in pymobiledevice3/services/mobile_activation.py 
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
async def send_command(self, command: str, value: str = ""):
    """
    Send a single command to the activation daemon over a fresh service connection.

    Opens the ``com.apple.mobileactivationd`` service, sends the command (with an optional
    ``Value``), and returns the response. The connection is closed afterwards.

    :param command: the activation daemon command name.
    :param value: optional value sent alongside the command; omitted when empty.
    :returns: the daemon's response plist.
    """
    data = {"Command": command}
    if value:
        data["Value"] = value
    async with _aclosing(await self.lockdown.start_lockdown_service(self.SERVICE_NAME)) as service:
        return await service.send_recv_plist(data)

post 

post(url: str, data: dict, headers: Optional[CaseInsensitiveDict[str, str]] = None) -> tuple[bytes, CaseInsensitiveDict[str]]

Perform an HTTP POST to an activation server endpoint.

Parameters:

Name Type Description Default
url str

the activation endpoint to post to.

 required
data dict

the form data or request body to send.

 required
headers Optional[CaseInsensitiveDict[str, str]]

optional request headers; the module's default activation headers are used when omitted.

 None

Returns:

Type Description
tuple[bytes, CaseInsensitiveDict[str]]

a tuple of the response body bytes and the response headers.
Source code in pymobiledevice3/services/mobile_activation.py 
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
def post(
    self, url: str, data: dict, headers: Optional[CaseInsensitiveDict[str, str]] = None
) -> tuple[bytes, CaseInsensitiveDict[str]]:
    """
    Perform an HTTP POST to an activation server endpoint.

    :param url: the activation endpoint to post to.
    :param data: the form data or request body to send.
    :param headers: optional request headers; the module's default activation headers are used
        when omitted.
    :returns: a tuple of the response body bytes and the response headers.
    """
    if headers is None:
        headers = DEFAULT_HEADERS

    resp = requests.post(url, data=data, headers=headers)
    return resp.content, resp.headers

pymobiledevice3.services.notification_proxy.NotificationProxyService

Bases: LockdownService

Post and observe Darwin notifications on the device via the notification proxy lockdown service.

Allows sending notifications to the device, registering interest in notifications so the device relays them back, and iterating over the relayed notifications. A secure or insecure variant of the service is selected by the insecure flag, and the RSD/tunnel variant is chosen automatically for RemoteServiceDiscoveryService providers. This is a lockdown service and is used as an async context manager.

Source code in pymobiledevice3/services/notification_proxy.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
class NotificationProxyService(LockdownService):
    """
    Post and observe Darwin notifications on the device via the notification proxy lockdown service.

    Allows sending notifications to the device, registering interest in notifications so the device
    relays them back, and iterating over the relayed notifications. A secure or insecure variant of
    the service is selected by the ``insecure`` flag, and the RSD/tunnel variant is chosen
    automatically for `RemoteServiceDiscoveryService` providers. This is a lockdown service and is
    used as an async context manager.
    """

    SERVICE_NAME = "com.apple.mobile.notification_proxy"
    RSD_SERVICE_NAME = "com.apple.mobile.notification_proxy.shim.remote"

    INSECURE_SERVICE_NAME = "com.apple.mobile.insecure_notification_proxy"
    RSD_INSECURE_SERVICE_NAME = "com.apple.mobile.insecure_notification_proxy.shim.remote"

    def __init__(self, lockdown: LockdownServiceProvider, insecure=False, timeout: Optional[Union[float, int]] = None):
        """
        :param lockdown: service provider used to start the service and reach the device.
        :param insecure: when True, use the insecure notification proxy service instead of the secure one.
        :param timeout: optional socket receive timeout in seconds applied to the service connection.
        """
        if isinstance(lockdown, RemoteServiceDiscoveryService):
            secure_service_name = self.RSD_SERVICE_NAME
            insecure_service_name = self.RSD_INSECURE_SERVICE_NAME
        else:
            secure_service_name = self.SERVICE_NAME
            insecure_service_name = self.INSECURE_SERVICE_NAME

        if insecure:
            super().__init__(lockdown, insecure_service_name)
        else:
            super().__init__(lockdown, secure_service_name)

        if timeout is not None:
            self.service.socket.settimeout(timeout)

    async def notify_post(self, name: str) -> None:
        """
        Post a notification on the device.

        Sends a ``PostNotification`` command, causing the device to broadcast the named notification.

        :param name: notification name to post (e.g. a Darwin notification name).
        """
        await self.service.send_plist({"Command": "PostNotification", "Name": name})

    async def notify_register_dispatch(self, name: str) -> None:
        """
        Register interest in a notification so the device relays it back.

        Sends an ``ObserveNotification`` command; once registered, the device sends a message
        whenever the named notification fires, which can be read via `receive_notification`.

        :param name: notification name to observe.
        """
        self.logger.info(f"Observing {name}")
        await self.service.send_plist({"Command": "ObserveNotification", "Name": name})

    async def receive_notification(self) -> AsyncGenerator[dict, None]:
        """
        Yield notifications relayed from the device for previously observed names.

        Continuously reads from the service and yields each received message until the connection
        is closed.

        :returns: an async generator of the received notification plists.
        :raises NotificationTimeoutError: if no notification arrives within the configured socket timeout.
        """
        while True:
            try:
                yield await self.service.recv_plist()
            except socket.timeout as e:
                raise NotificationTimeoutError from e

notify_post async 

notify_post(name: str) -> None

Post a notification on the device.

Sends a PostNotification command, causing the device to broadcast the named notification.

Parameters:

Name Type Description Default
name str

notification name to post (e.g. a Darwin notification name).

 required
Source code in pymobiledevice3/services/notification_proxy.py 
49
50
51
52
53
54
55
56
57
async def notify_post(self, name: str) -> None:
    """
    Post a notification on the device.

    Sends a ``PostNotification`` command, causing the device to broadcast the named notification.

    :param name: notification name to post (e.g. a Darwin notification name).
    """
    await self.service.send_plist({"Command": "PostNotification", "Name": name})

notify_register_dispatch async 

notify_register_dispatch(name: str) -> None

Register interest in a notification so the device relays it back.

Sends an ObserveNotification command; once registered, the device sends a message whenever the named notification fires, which can be read via receive_notification.

Parameters:

Name Type Description Default
name str

notification name to observe.

 required
Source code in pymobiledevice3/services/notification_proxy.py 
59
60
61
62
63
64
65
66
67
68
69
async def notify_register_dispatch(self, name: str) -> None:
    """
    Register interest in a notification so the device relays it back.

    Sends an ``ObserveNotification`` command; once registered, the device sends a message
    whenever the named notification fires, which can be read via `receive_notification`.

    :param name: notification name to observe.
    """
    self.logger.info(f"Observing {name}")
    await self.service.send_plist({"Command": "ObserveNotification", "Name": name})

receive_notification async 

receive_notification() -> AsyncGenerator[dict, None]

Yield notifications relayed from the device for previously observed names.

Continuously reads from the service and yields each received message until the connection is closed.

Returns:

Type Description
AsyncGenerator[dict, None]

an async generator of the received notification plists.

Raises:

Type Description
NotificationTimeoutError

if no notification arrives within the configured socket timeout.
Source code in pymobiledevice3/services/notification_proxy.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
async def receive_notification(self) -> AsyncGenerator[dict, None]:
    """
    Yield notifications relayed from the device for previously observed names.

    Continuously reads from the service and yields each received message until the connection
    is closed.

    :returns: an async generator of the received notification plists.
    :raises NotificationTimeoutError: if no notification arrives within the configured socket timeout.
    """
    while True:
        try:
            yield await self.service.recv_plist()
        except socket.timeout as e:
            raise NotificationTimeoutError from e

pymobiledevice3.services.power_assertion.PowerAssertionService

Bases: LockdownService

Hold an IOKit power assertion on the device through the assertion agent lockdown service.

A power assertion prevents the device from sleeping (or otherwise alters its power behavior) for as long as it is held. This is a lockdown service and is used as an async context manager; the RSD/tunnel variant is selected automatically for non-LockdownClient providers.

Source code in pymobiledevice3/services/power_assertion.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
class PowerAssertionService(LockdownService):
    """
    Hold an IOKit power assertion on the device through the assertion agent lockdown service.

    A power assertion prevents the device from sleeping (or otherwise alters its power behavior)
    for as long as it is held. This is a lockdown service and is used as an async context manager;
    the RSD/tunnel variant is selected automatically for non-`LockdownClient` providers.
    """

    RSD_SERVICE_NAME = "com.apple.mobile.assertion_agent.shim.remote"
    SERVICE_NAME = "com.apple.mobile.assertion_agent"

    def __init__(self, lockdown: LockdownServiceProvider):
        if isinstance(lockdown, LockdownClient):
            super().__init__(lockdown, self.SERVICE_NAME)
        else:
            super().__init__(lockdown, self.RSD_SERVICE_NAME)

    @contextlib.asynccontextmanager
    async def create_power_assertion(self, type_: str, name: str, timeout: float, details: Optional[str] = None):
        """
        Create a power assertion on the device, held for the duration of the ``with`` block.

        Sends a create-assertion command to the agent, which calls ``IOPMAssertionCreateWithName``
        on the device. This is an async context manager: the assertion stays active while the block
        is open.

        :param type_: IOKit assertion type (e.g. the assertion-type string passed to
            ``IOPMAssertionCreateWithName``).
        :param name: human-readable name identifying the assertion.
        :param timeout: assertion timeout in seconds, after which the device may release it.
        :param details: optional descriptive detail string attached to the assertion.
        """
        msg = {
            "CommandKey": "CommandCreateAssertion",
            "AssertionTypeKey": type_,
            "AssertionNameKey": name,
            "AssertionTimeoutKey": timeout,
        }

        if details is not None:
            msg["AssertionDetailKey"] = details

        await self.service.send_recv_plist(msg)
        yield

create_power_assertion async 

create_power_assertion(type_: str, name: str, timeout: float, details: Optional[str] = None)

Create a power assertion on the device, held for the duration of the with block.

Sends a create-assertion command to the agent, which calls IOPMAssertionCreateWithName on the device. This is an async context manager: the assertion stays active while the block is open.

Parameters:

Name Type Description Default
type_ str

IOKit assertion type (e.g. the assertion-type string passed to IOPMAssertionCreateWithName).

 required
name str

human-readable name identifying the assertion.

 required
timeout float

assertion timeout in seconds, after which the device may release it.

 required
details Optional[str]

optional descriptive detail string attached to the assertion.

 None
Source code in pymobiledevice3/services/power_assertion.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
@contextlib.asynccontextmanager
async def create_power_assertion(self, type_: str, name: str, timeout: float, details: Optional[str] = None):
    """
    Create a power assertion on the device, held for the duration of the ``with`` block.

    Sends a create-assertion command to the agent, which calls ``IOPMAssertionCreateWithName``
    on the device. This is an async context manager: the assertion stays active while the block
    is open.

    :param type_: IOKit assertion type (e.g. the assertion-type string passed to
        ``IOPMAssertionCreateWithName``).
    :param name: human-readable name identifying the assertion.
    :param timeout: assertion timeout in seconds, after which the device may release it.
    :param details: optional descriptive detail string attached to the assertion.
    """
    msg = {
        "CommandKey": "CommandCreateAssertion",
        "AssertionTypeKey": type_,
        "AssertionNameKey": name,
        "AssertionTimeoutKey": timeout,
    }

    if details is not None:
        msg["AssertionDetailKey"] = details

    await self.service.send_recv_plist(msg)
    yield

pymobiledevice3.services.heartbeat.HeartbeatService

Keep an active connection alive with the device's heartbeat lockdown service.

The device periodically sends Marco messages over com.apple.mobile.heartbeat and expects a Polo reply; exchanging them prevents the connection from being torn down. The RSD/tunnel variant of the service is selected automatically when the provider is not a plain LockdownClient.

Source code in pymobiledevice3/services/heartbeat.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
class HeartbeatService:
    """
    Keep an active connection alive with the device's heartbeat lockdown service.

    The device periodically sends ``Marco`` messages over ``com.apple.mobile.heartbeat`` and expects a
    ``Polo`` reply; exchanging them prevents the connection from being torn down. The RSD/tunnel variant
    of the service is selected automatically when the provider is not a plain `LockdownClient`.
    """

    SERVICE_NAME = "com.apple.mobile.heartbeat"
    RSD_SERVICE_NAME = "com.apple.mobile.heartbeat.shim.remote"

    def __init__(self, lockdown: LockdownServiceProvider):
        self.logger = logging.getLogger(__name__)
        self.lockdown = lockdown

        if isinstance(lockdown, LockdownClient):
            self.service_name = self.SERVICE_NAME
        else:
            self.service_name = self.RSD_SERVICE_NAME

    async def start(self, interval: Optional[float] = None) -> None:
        """
        Start the heartbeat exchange loop.

        Opens the heartbeat service and, in a loop, receives each message from the device and replies
        with ``Polo`` to keep the connection alive.

        :param interval: when set, stop the loop once roughly this many seconds have elapsed since it
            started; when None, the loop runs indefinitely.
        """
        start = time.time()
        service = await self.lockdown.start_lockdown_service(self.service_name)

        while True:
            response = await service.recv_plist()
            self.logger.debug(response)

            if interval is not None and time.time() >= start + interval:
                break

            await service.send_plist({"Command": "Polo"})

start async 

start(interval: Optional[float] = None) -> None

Start the heartbeat exchange loop.

Opens the heartbeat service and, in a loop, receives each message from the device and replies with Polo to keep the connection alive.

Parameters:

Name Type Description Default
interval Optional[float]

when set, stop the loop once roughly this many seconds have elapsed since it started; when None, the loop runs indefinitely.

 None
Source code in pymobiledevice3/services/heartbeat.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
async def start(self, interval: Optional[float] = None) -> None:
    """
    Start the heartbeat exchange loop.

    Opens the heartbeat service and, in a loop, receives each message from the device and replies
    with ``Polo`` to keep the connection alive.

    :param interval: when set, stop the loop once roughly this many seconds have elapsed since it
        started; when None, the loop runs indefinitely.
    """
    start = time.time()
    service = await self.lockdown.start_lockdown_service(self.service_name)

    while True:
        response = await service.recv_plist()
        self.logger.debug(response)

        if interval is not None and time.time() >= start + interval:
            break

        await service.send_plist({"Command": "Polo"})

pymobiledevice3.services.companion.CompanionProxyService

Bases: LockdownService

Query and interact with companion devices (e.g. a paired Apple Watch) via the companion proxy lockdown service.

Provides access to the paired-device registry, live notifications of companion devices coming and going, registry value lookups, and TCP port forwarding to a companion device. This is a lockdown service and is used as an async context manager; the RSD/tunnel variant is selected automatically for non-LockdownClient providers.

Source code in pymobiledevice3/services/companion.py 
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
class CompanionProxyService(LockdownService):
    """
    Query and interact with companion devices (e.g. a paired Apple Watch) via the companion proxy
    lockdown service.

    Provides access to the paired-device registry, live notifications of companion devices coming
    and going, registry value lookups, and TCP port forwarding to a companion device. This is a
    lockdown service and is used as an async context manager; the RSD/tunnel variant is selected
    automatically for non-`LockdownClient` providers.
    """

    SERVICE_NAME = "com.apple.companion_proxy"
    RSD_SERVICE_NAME = "com.apple.companion_proxy.shim.remote"

    def __init__(self, lockdown: LockdownServiceProvider):
        if isinstance(lockdown, LockdownClient):
            super().__init__(lockdown, self.SERVICE_NAME)
        else:
            super().__init__(lockdown, self.RSD_SERVICE_NAME)

    async def list(self):
        """
        List the companion devices currently paired with the device.

        Sends a ``GetDeviceRegistry`` command and returns the paired-device registry.

        :returns: the list of paired devices (the ``PairedDevicesArray`` entry), or an empty list
            if none are present.
        """
        service = await self.lockdown.start_lockdown_service(self.service_name)
        return (await service.send_recv_plist({"Command": "GetDeviceRegistry"})).get("PairedDevicesArray", [])

    async def listen_for_devices(self):
        """
        Yield events as companion devices appear and disappear.

        Sends ``StartListeningForDevices`` and then continuously yields each event message the
        device emits.

        :returns: an async generator of device registry change events.
        """
        service = await self.lockdown.start_lockdown_service(self.service_name)
        await service.send_plist({"Command": "StartListeningForDevices"})
        while True:
            yield await service.recv_plist()

    async def get_value(self, udid: str, key: str):
        """
        Read a single value from the registry of a specific companion device.

        Sends a ``GetValueFromRegistry`` command for the given device and key.

        :param udid: UDID of the companion device to query.
        :param key: registry key whose value should be retrieved.
        :returns: the retrieved value dictionary.
        :raises PyMobileDevice3Exception: if the device returns an error instead of a value.
        """
        service = await self.lockdown.start_lockdown_service(self.service_name)
        response = await service.send_recv_plist({
            "Command": "GetValueFromRegistry",
            "GetValueGizmoUDIDKey": udid,
            "GetValueKeyKey": key,
        })

        value = response.get("RetrievedValueDictionary")
        if value is not None:
            return value

        error = response.get("Error")
        raise PyMobileDevice3Exception(error)

    async def start_forwarding_service_port(
        self, remote_port: int, service_name: Optional[str] = None, options: Optional[dict] = None
    ):
        """
        Start forwarding a port on the companion device through the proxy.

        Sends a ``StartForwardingServicePort`` command for the given remote port. The request
        defaults the forwarded service to non-low-priority and does not prefer Wi-Fi; these and any
        other fields may be overridden via ``options``.

        :param remote_port: port number on the companion device to forward.
        :param service_name: optional name of the forwarded service, sent as ``ForwardedServiceName``.
        :param options: optional dictionary merged into the request to override default fields.
        :returns: the local proxy port assigned for the forwarded connection
            (the ``CompanionProxyServicePort`` value).
        """
        service = await self.lockdown.start_lockdown_service(self.service_name)

        request = {
            "Command": "StartForwardingServicePort",
            "GizmoRemotePortNumber": remote_port,
            "IsServiceLowPriority": False,
            "PreferWifi": False,
        }

        if service_name is not None:
            request["ForwardedServiceName"] = service_name

        if options is not None:
            request.update(options)

        return (await service.send_recv_plist(request)).get("CompanionProxyServicePort")

    async def stop_forwarding_service_port(self, remote_port: int):
        """
        Stop a port forward previously started with `start_forwarding_service_port`.

        Sends a ``StopForwardingServicePort`` command for the given remote port.

        :param remote_port: port number on the companion device whose forwarding should be stopped.
        :returns: the device's response to the stop command.
        """
        service = await self.lockdown.start_lockdown_service(self.service_name)

        request = {"Command": "StopForwardingServicePort", "GizmoRemotePortNumber": remote_port}

        return await service.send_recv_plist(request)

list async 

list()

List the companion devices currently paired with the device.

Sends a GetDeviceRegistry command and returns the paired-device registry.

Returns:

Type Description

the list of paired devices (the PairedDevicesArray entry), or an empty list if none are present.
Source code in pymobiledevice3/services/companion.py 
30
31
32
33
34
35
36
37
38
39
40
async def list(self):
    """
    List the companion devices currently paired with the device.

    Sends a ``GetDeviceRegistry`` command and returns the paired-device registry.

    :returns: the list of paired devices (the ``PairedDevicesArray`` entry), or an empty list
        if none are present.
    """
    service = await self.lockdown.start_lockdown_service(self.service_name)
    return (await service.send_recv_plist({"Command": "GetDeviceRegistry"})).get("PairedDevicesArray", [])

listen_for_devices async 

listen_for_devices()

Yield events as companion devices appear and disappear.

Sends StartListeningForDevices and then continuously yields each event message the device emits.

Returns:

Type Description

an async generator of device registry change events.
Source code in pymobiledevice3/services/companion.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
async def listen_for_devices(self):
    """
    Yield events as companion devices appear and disappear.

    Sends ``StartListeningForDevices`` and then continuously yields each event message the
    device emits.

    :returns: an async generator of device registry change events.
    """
    service = await self.lockdown.start_lockdown_service(self.service_name)
    await service.send_plist({"Command": "StartListeningForDevices"})
    while True:
        yield await service.recv_plist()

get_value async 

get_value(udid: str, key: str)

Read a single value from the registry of a specific companion device.

Sends a GetValueFromRegistry command for the given device and key.

Parameters:

Name Type Description Default
udid str

UDID of the companion device to query.

 required
key str

registry key whose value should be retrieved.

 required

Returns:

Type Description

the retrieved value dictionary.

Raises:

Type Description
PyMobileDevice3Exception

if the device returns an error instead of a value.
Source code in pymobiledevice3/services/companion.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
async def get_value(self, udid: str, key: str):
    """
    Read a single value from the registry of a specific companion device.

    Sends a ``GetValueFromRegistry`` command for the given device and key.

    :param udid: UDID of the companion device to query.
    :param key: registry key whose value should be retrieved.
    :returns: the retrieved value dictionary.
    :raises PyMobileDevice3Exception: if the device returns an error instead of a value.
    """
    service = await self.lockdown.start_lockdown_service(self.service_name)
    response = await service.send_recv_plist({
        "Command": "GetValueFromRegistry",
        "GetValueGizmoUDIDKey": udid,
        "GetValueKeyKey": key,
    })

    value = response.get("RetrievedValueDictionary")
    if value is not None:
        return value

    error = response.get("Error")
    raise PyMobileDevice3Exception(error)

start_forwarding_service_port async 

start_forwarding_service_port(remote_port: int, service_name: Optional[str] = None, options: Optional[dict] = None)

Start forwarding a port on the companion device through the proxy.

Sends a StartForwardingServicePort command for the given remote port. The request defaults the forwarded service to non-low-priority and does not prefer Wi-Fi; these and any other fields may be overridden via options.

Parameters:

Name Type Description Default
remote_port int

port number on the companion device to forward.

 required
service_name Optional[str]

optional name of the forwarded service, sent as ForwardedServiceName.

 None
options Optional[dict]

optional dictionary merged into the request to override default fields.

 None

Returns:

Type Description

the local proxy port assigned for the forwarded connection (the CompanionProxyServicePort value).
Source code in pymobiledevice3/services/companion.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
async def start_forwarding_service_port(
    self, remote_port: int, service_name: Optional[str] = None, options: Optional[dict] = None
):
    """
    Start forwarding a port on the companion device through the proxy.

    Sends a ``StartForwardingServicePort`` command for the given remote port. The request
    defaults the forwarded service to non-low-priority and does not prefer Wi-Fi; these and any
    other fields may be overridden via ``options``.

    :param remote_port: port number on the companion device to forward.
    :param service_name: optional name of the forwarded service, sent as ``ForwardedServiceName``.
    :param options: optional dictionary merged into the request to override default fields.
    :returns: the local proxy port assigned for the forwarded connection
        (the ``CompanionProxyServicePort`` value).
    """
    service = await self.lockdown.start_lockdown_service(self.service_name)

    request = {
        "Command": "StartForwardingServicePort",
        "GizmoRemotePortNumber": remote_port,
        "IsServiceLowPriority": False,
        "PreferWifi": False,
    }

    if service_name is not None:
        request["ForwardedServiceName"] = service_name

    if options is not None:
        request.update(options)

    return (await service.send_recv_plist(request)).get("CompanionProxyServicePort")

stop_forwarding_service_port async 

stop_forwarding_service_port(remote_port: int)

Stop a port forward previously started with start_forwarding_service_port.

Sends a StopForwardingServicePort command for the given remote port.

Parameters:

Name Type Description Default
remote_port int

port number on the companion device whose forwarding should be stopped.

 required

Returns:

Type Description

the device's response to the stop command.
Source code in pymobiledevice3/services/companion.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
async def stop_forwarding_service_port(self, remote_port: int):
    """
    Stop a port forward previously started with `start_forwarding_service_port`.

    Sends a ``StopForwardingServicePort`` command for the given remote port.

    :param remote_port: port number on the companion device whose forwarding should be stopped.
    :returns: the device's response to the stop command.
    """
    service = await self.lockdown.start_lockdown_service(self.service_name)

    request = {"Command": "StopForwardingServicePort", "GizmoRemotePortNumber": remote_port}

    return await service.send_recv_plist(request)

pymobiledevice3.services.preboard.PreboardService

Bases: LockdownService

Manage preboard stashbags on the device via the com.apple.preboardservice_v2 lockdown service.

A stashbag holds key material that allows the device to unlock data-protection classes during an unattended reboot (e.g. before the user enters their passcode). This service creates and commits such stashbags. This is a lockdown service and is used as an async context manager; the RSD/tunnel variant is selected automatically for non-LockdownClient providers.

Source code in pymobiledevice3/services/preboard.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
class PreboardService(LockdownService):
    """
    Manage preboard stashbags on the device via the ``com.apple.preboardservice_v2`` lockdown service.

    A stashbag holds key material that allows the device to unlock data-protection classes during an
    unattended reboot (e.g. before the user enters their passcode). This service creates and commits
    such stashbags. This is a lockdown service and is used as an async context manager; the
    RSD/tunnel variant is selected automatically for non-`LockdownClient` providers.
    """

    RSD_SERVICE_NAME = "com.apple.preboardservice_v2.shim.remote"
    SERVICE_NAME = "com.apple.preboardservice_v2"

    def __init__(self, lockdown: LockdownServiceProvider):
        if isinstance(lockdown, LockdownClient):
            super().__init__(lockdown, self.SERVICE_NAME)
        else:
            super().__init__(lockdown, self.RSD_SERVICE_NAME)

    async def create_stashbag(self, manifest):
        """
        Create a stashbag on the device.

        Sends a ``CreateStashbag`` command with the given manifest.

        :param manifest: the stashbag manifest describing the key material to stash.
        :returns: the device's response to the command.
        """
        return await self.service.send_recv_plist({"Command": "CreateStashbag", "Manifest": manifest})

    async def commit(self, manifest):
        """
        Commit a previously created stashbag on the device.

        Sends a ``CommitStashbag`` command with the given manifest.

        :param manifest: the stashbag manifest to commit.
        :returns: the device's response to the command.
        """
        return await self.service.send_recv_plist({"Command": "CommitStashbag", "Manifest": manifest})

create_stashbag async 

create_stashbag(manifest)

Create a stashbag on the device.

Sends a CreateStashbag command with the given manifest.

Parameters:

Name Type Description Default
manifest

the stashbag manifest describing the key material to stash.

 required

Returns:

Type Description

the device's response to the command.
Source code in pymobiledevice3/services/preboard.py 
27
28
29
30
31
32
33
34
35
36
async def create_stashbag(self, manifest):
    """
    Create a stashbag on the device.

    Sends a ``CreateStashbag`` command with the given manifest.

    :param manifest: the stashbag manifest describing the key material to stash.
    :returns: the device's response to the command.
    """
    return await self.service.send_recv_plist({"Command": "CreateStashbag", "Manifest": manifest})

commit async 

commit(manifest)

Commit a previously created stashbag on the device.

Sends a CommitStashbag command with the given manifest.

Parameters:

Name Type Description Default
manifest

the stashbag manifest to commit.

 required

Returns:

Type Description

the device's response to the command.
Source code in pymobiledevice3/services/preboard.py 
38
39
40
41
42
43
44
45
46
47
async def commit(self, manifest):
    """
    Commit a previously created stashbag on the device.

    Sends a ``CommitStashbag`` command with the given manifest.

    :param manifest: the stashbag manifest to commit.
    :returns: the device's response to the command.
    """
    return await self.service.send_recv_plist({"Command": "CommitStashbag", "Manifest": manifest})

pymobiledevice3.services.restore_service.RestoreService

Bases: RemoteService

Issue restore-time control commands to restoreserviced over RemoteXPC.

Wraps the com.apple.RestoreRemoteServices.restoreserviced remote service, exposing commands used during the restore/recovery flow such as entering recovery, rebooting, and reading nonces and preflight information. This is a remote service reached over RSD and is used as an async context manager.

Source code in pymobiledevice3/services/restore_service.py 
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
class RestoreService(RemoteService):
    """
    Issue restore-time control commands to ``restoreserviced`` over RemoteXPC.

    Wraps the ``com.apple.RestoreRemoteServices.restoreserviced`` remote service, exposing commands
    used during the restore/recovery flow such as entering recovery, rebooting, and reading nonces
    and preflight information. This is a remote service reached over RSD and is used as an async
    context manager.
    """

    SERVICE_NAME = "com.apple.RestoreRemoteServices.restoreserviced"

    def __init__(self, lockdown: RemoteServiceDiscoveryService):
        super().__init__(lockdown, self.SERVICE_NAME)

    async def delay_recovery_image(self) -> None:
        """
        Request that the recovery image be delayed.

        Sends the ``delayrecoveryimage`` command, which is only honored on devices of ProductType
        ``0x1677b394`` and otherwise fails.

        :raises PyMobileDevice3Exception: if the device does not report success.
        """
        await self.validate_command("delayrecoveryimage")

    async def enter_recovery(self) -> None:
        """
        Put the device into recovery mode.

        Sends the ``recovery`` command.

        :raises PyMobileDevice3Exception: if the device does not report success.
        """
        await self.validate_command("recovery")

    async def reboot(self) -> None:
        """
        Reboot the device.

        Sends the ``reboot`` command.

        :raises PyMobileDevice3Exception: if the device does not report success.
        """
        await self.validate_command("reboot")

    async def get_preflightinfo(self) -> dict:
        """
        Retrieve restore preflight information from the device.

        Sends the ``getpreflightinfo`` command.

        :returns: the device's preflight info response.
        """
        return await self.service.send_receive_request({"command": "getpreflightinfo"})

    async def get_nonces(self) -> dict:
        """
        Retrieve the device's restore nonces.

        Sends the ``getnonces`` command.

        :returns: the response containing the ApNonce and SEPNonce.
        """
        return await self.service.send_receive_request({"command": "getnonces"})

    async def get_app_parameters(self) -> dict:
        """
        Retrieve restore app parameters from the device.

        Sends the ``getappparameters`` command.

        :returns: the device's app parameters response.
        :raises PyMobileDevice3Exception: if the device does not report success.
        """
        return await self.validate_command("getappparameters")

    async def restore_lang(self, language: str) -> dict:
        """
        Set the restore language.

        Sends the ``restorelang`` command with the given language as its argument.

        :param language: the language identifier to set for the restore.
        :returns: the device's response to the command.
        """
        return await self.service.send_receive_request({"command": "restorelang", "argument": language})

    async def validate_command(self, command: str) -> dict:
        """
        Send a command and assert that the device reports success.

        Sends the given command and verifies the response ``result`` is ``"success"``.

        :param command: the restore command name to send.
        :returns: the device's response.
        :raises PyMobileDevice3Exception: if the response result is not ``"success"``.
        """
        response = await self.service.send_receive_request({"command": command})
        if response.get("result") != "success":
            raise PyMobileDevice3Exception(f"request command: {command} failed with error: {response}")
        return response

delay_recovery_image async 

delay_recovery_image() -> None

Request that the recovery image be delayed.

Sends the delayrecoveryimage command, which is only honored on devices of ProductType 0x1677b394 and otherwise fails.

Raises:

Type Description
PyMobileDevice3Exception

if the device does not report success.
Source code in pymobiledevice3/services/restore_service.py 
21
22
23
24
25
26
27
28
29
30
async def delay_recovery_image(self) -> None:
    """
    Request that the recovery image be delayed.

    Sends the ``delayrecoveryimage`` command, which is only honored on devices of ProductType
    ``0x1677b394`` and otherwise fails.

    :raises PyMobileDevice3Exception: if the device does not report success.
    """
    await self.validate_command("delayrecoveryimage")

enter_recovery async 

enter_recovery() -> None

Put the device into recovery mode.

Sends the recovery command.

Raises:

Type Description
PyMobileDevice3Exception

if the device does not report success.
Source code in pymobiledevice3/services/restore_service.py 
32
33
34
35
36
37
38
39
40
async def enter_recovery(self) -> None:
    """
    Put the device into recovery mode.

    Sends the ``recovery`` command.

    :raises PyMobileDevice3Exception: if the device does not report success.
    """
    await self.validate_command("recovery")

reboot async 

reboot() -> None

Reboot the device.

Sends the reboot command.

Raises:

Type Description
PyMobileDevice3Exception

if the device does not report success.
Source code in pymobiledevice3/services/restore_service.py 
42
43
44
45
46
47
48
49
50
async def reboot(self) -> None:
    """
    Reboot the device.

    Sends the ``reboot`` command.

    :raises PyMobileDevice3Exception: if the device does not report success.
    """
    await self.validate_command("reboot")

get_preflightinfo async 

get_preflightinfo() -> dict

Retrieve restore preflight information from the device.

Sends the getpreflightinfo command.

Returns:

Type Description
dict

the device's preflight info response.
Source code in pymobiledevice3/services/restore_service.py 
52
53
54
55
56
57
58
59
60
async def get_preflightinfo(self) -> dict:
    """
    Retrieve restore preflight information from the device.

    Sends the ``getpreflightinfo`` command.

    :returns: the device's preflight info response.
    """
    return await self.service.send_receive_request({"command": "getpreflightinfo"})

get_nonces async 

get_nonces() -> dict

Retrieve the device's restore nonces.

Sends the getnonces command.

Returns:

Type Description
dict

the response containing the ApNonce and SEPNonce.
Source code in pymobiledevice3/services/restore_service.py 
62
63
64
65
66
67
68
69
70
async def get_nonces(self) -> dict:
    """
    Retrieve the device's restore nonces.

    Sends the ``getnonces`` command.

    :returns: the response containing the ApNonce and SEPNonce.
    """
    return await self.service.send_receive_request({"command": "getnonces"})

get_app_parameters async 

get_app_parameters() -> dict

Retrieve restore app parameters from the device.

Sends the getappparameters command.

Returns:

Type Description
dict

the device's app parameters response.

Raises:

Type Description
PyMobileDevice3Exception

if the device does not report success.
Source code in pymobiledevice3/services/restore_service.py 
72
73
74
75
76
77
78
79
80
81
async def get_app_parameters(self) -> dict:
    """
    Retrieve restore app parameters from the device.

    Sends the ``getappparameters`` command.

    :returns: the device's app parameters response.
    :raises PyMobileDevice3Exception: if the device does not report success.
    """
    return await self.validate_command("getappparameters")

restore_lang async 

restore_lang(language: str) -> dict

Set the restore language.

Sends the restorelang command with the given language as its argument.

Parameters:

Name Type Description Default
language str

the language identifier to set for the restore.

 required

Returns:

Type Description
dict

the device's response to the command.
Source code in pymobiledevice3/services/restore_service.py 
83
84
85
86
87
88
89
90
91
92
async def restore_lang(self, language: str) -> dict:
    """
    Set the restore language.

    Sends the ``restorelang`` command with the given language as its argument.

    :param language: the language identifier to set for the restore.
    :returns: the device's response to the command.
    """
    return await self.service.send_receive_request({"command": "restorelang", "argument": language})

validate_command async 

validate_command(command: str) -> dict

Send a command and assert that the device reports success.

Sends the given command and verifies the response result is "success".

Parameters:

Name Type Description Default
command str

the restore command name to send.

 required

Returns:

Type Description
dict

the device's response.

Raises:

Type Description
PyMobileDevice3Exception

if the response result is not "success".
Source code in pymobiledevice3/services/restore_service.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
async def validate_command(self, command: str) -> dict:
    """
    Send a command and assert that the device reports success.

    Sends the given command and verifies the response ``result`` is ``"success"``.

    :param command: the restore command name to send.
    :returns: the device's response.
    :raises PyMobileDevice3Exception: if the response result is not ``"success"``.
    """
    response = await self.service.send_receive_request({"command": command})
    if response.get("result") != "success":
        raise PyMobileDevice3Exception(f"request command: {command} failed with error: {response}")
    return response

pymobiledevice3.services.device_arbitration.DtDeviceArbitration

Bases: LockdownService

Claim and release exclusive use of a device via the com.apple.dt.devicearbitration service.

Device arbitration lets a host check a device in (claiming it, optionally forcibly) so other hosts know it is in use, and check it back out when done. This is a developer lockdown service (it requires the DeveloperDiskImage to be mounted) and is used as an async context manager.

Source code in pymobiledevice3/services/device_arbitration.py 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
class DtDeviceArbitration(LockdownService):
    """
    Claim and release exclusive use of a device via the ``com.apple.dt.devicearbitration`` service.

    Device arbitration lets a host check a device in (claiming it, optionally forcibly) so other
    hosts know it is in use, and check it back out when done. This is a developer lockdown service
    (it requires the DeveloperDiskImage to be mounted) and is used as an async context manager.
    """

    SERVICE_NAME = "com.apple.dt.devicearbitration"

    def __init__(self, lockdown: LockdownClient):
        super().__init__(lockdown, self.SERVICE_NAME, is_developer_service=True)

    async def _send_recv(self, request: dict) -> dict:
        return await self.service.send_recv_plist(request)

    async def version(self) -> dict:
        """
        Query the arbitration service version.

        Sends the ``version`` command.

        :returns: the device's version response.
        """
        return await self._send_recv({"command": "version"})

    async def check_in(self, hostname: str, force: bool = False):
        """
        Claim the device for the given host.

        Sends a ``check-in`` command tagging the device with ``hostname``. When ``force`` is True,
        a ``force-check-in`` command is sent instead, taking the device even if another host holds it.

        :param hostname: identifier of the host claiming the device.
        :param force: when True, forcibly claim the device even if it is already in use.
        :raises DeviceAlreadyInUseError: if the device is already checked in by another host
            (and ``force`` is not used).
        """
        request = {"command": "check-in", "hostname": hostname}
        if force:
            request["command"] = "force-check-in"
        response = await self._send_recv(request)
        if response.get("result") != "success":
            raise DeviceAlreadyInUseError(response)

    async def check_out(self):
        """
        Release a previously claimed device.

        Sends the ``check-out`` command.

        :raises ArbitrationError: if the device does not report success.
        """
        request = {"command": "check-out"}
        response = await self._send_recv(request)
        if response.get("result") != "success":
            raise ArbitrationError(f"failed with: {response}")

version async 

version() -> dict

Query the arbitration service version.

Sends the version command.

Returns:

Type Description
dict

the device's version response.
Source code in pymobiledevice3/services/device_arbitration.py 
23
24
25
26
27
28
29
30
31
async def version(self) -> dict:
    """
    Query the arbitration service version.

    Sends the ``version`` command.

    :returns: the device's version response.
    """
    return await self._send_recv({"command": "version"})

check_in async 

check_in(hostname: str, force: bool = False)

Claim the device for the given host.

Sends a check-in command tagging the device with hostname. When force is True, a force-check-in command is sent instead, taking the device even if another host holds it.

Parameters:

Name Type Description Default
hostname str

identifier of the host claiming the device.

 required
force bool

when True, forcibly claim the device even if it is already in use.

 False

Raises:

Type Description
DeviceAlreadyInUseError

if the device is already checked in by another host (and force is not used).
Source code in pymobiledevice3/services/device_arbitration.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
async def check_in(self, hostname: str, force: bool = False):
    """
    Claim the device for the given host.

    Sends a ``check-in`` command tagging the device with ``hostname``. When ``force`` is True,
    a ``force-check-in`` command is sent instead, taking the device even if another host holds it.

    :param hostname: identifier of the host claiming the device.
    :param force: when True, forcibly claim the device even if it is already in use.
    :raises DeviceAlreadyInUseError: if the device is already checked in by another host
        (and ``force`` is not used).
    """
    request = {"command": "check-in", "hostname": hostname}
    if force:
        request["command"] = "force-check-in"
    response = await self._send_recv(request)
    if response.get("result") != "success":
        raise DeviceAlreadyInUseError(response)

check_out async 

check_out()

Release a previously claimed device.

Sends the check-out command.

Raises:

Type Description
ArbitrationError

if the device does not report success.
Source code in pymobiledevice3/services/device_arbitration.py 
52
53
54
55
56
57
58
59
60
61
62
63
async def check_out(self):
    """
    Release a previously claimed device.

    Sends the ``check-out`` command.

    :raises ArbitrationError: if the device does not report success.
    """
    request = {"command": "check-out"}
    response = await self._send_recv(request)
    if response.get("result") != "success":
        raise ArbitrationError(f"failed with: {response}")

pymobiledevice3.services.idam.IDAMService

Bases: LockdownService

Query and set the device's IDAM configuration via the com.apple.idamd lockdown service.

Exposes reading the current IDAM configuration and toggling it. This is a lockdown service and is used as an async context manager; the RSD/tunnel variant is selected automatically for non-LockdownClient providers.

Source code in pymobiledevice3/services/idam.py 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
class IDAMService(LockdownService):
    """
    Query and set the device's IDAM configuration via the ``com.apple.idamd`` lockdown service.

    Exposes reading the current IDAM configuration and toggling it. This is a lockdown service and
    is used as an async context manager; the RSD/tunnel variant is selected automatically for
    non-`LockdownClient` providers.
    """

    RSD_SERVICE_NAME = "com.apple.idamd.shim.remote"
    SERVICE_NAME = "com.apple.idamd"

    def __init__(self, lockdown: LockdownServiceProvider) -> None:
        if isinstance(lockdown, LockdownClient):
            super().__init__(lockdown, self.SERVICE_NAME)
        else:
            super().__init__(lockdown, self.RSD_SERVICE_NAME)

    async def configuration_inquiry(self) -> dict:
        """
        Read the device's current IDAM configuration.

        Sends a ``Configuration Inquiry`` request.

        :returns: the device's IDAM configuration response.
        """
        return await self.service.send_recv_plist({"Configuration Inquiry": True})

    async def set_idam_configuration(self, value: bool) -> None:
        """
        Set the device's IDAM configuration.

        Sends a ``Set IDAM Configuration`` request with the given value.

        :param value: the IDAM configuration flag to apply.
        """
        await self.service.send_recv_plist({"Set IDAM Configuration": value})

configuration_inquiry async 

configuration_inquiry() -> dict

Read the device's current IDAM configuration.

Sends a Configuration Inquiry request.

Returns:

Type Description
dict

the device's IDAM configuration response.
Source code in pymobiledevice3/services/idam.py 
24
25
26
27
28
29
30
31
32
async def configuration_inquiry(self) -> dict:
    """
    Read the device's current IDAM configuration.

    Sends a ``Configuration Inquiry`` request.

    :returns: the device's IDAM configuration response.
    """
    return await self.service.send_recv_plist({"Configuration Inquiry": True})

set_idam_configuration async 

set_idam_configuration(value: bool) -> None

Set the device's IDAM configuration.

Sends a Set IDAM Configuration request with the given value.

Parameters:

Name Type Description Default
value bool

the IDAM configuration flag to apply.

 required
Source code in pymobiledevice3/services/idam.py 
34
35
36
37
38
39
40
41
42
async def set_idam_configuration(self, value: bool) -> None:
    """
    Set the device's IDAM configuration.

    Sends a ``Set IDAM Configuration`` request with the given value.

    :param value: the IDAM configuration flag to apply.
    """
    await self.service.send_recv_plist({"Set IDAM Configuration": value})