Add supervised configuration profile installs

pymobiledevice3/cli/profile.py

@@ -37,11 +37,25 @@ def profile_install(lockdown: LockdownClient, profiles):
         service.install_profile(profile.read())
 
 
+@profile_group.command('install-silent', cls=Command)
+@click.option('--keystore', type=click.File('rb'), required=True, help="A PKCS#12 keystore containing the certificate and private key which can supervise the device.")
+@click.option('--keystore-password', prompt=True, required=True, hide_input=True, help="The password for the PKCS#12 keystore.")
+@click.argument('profiles', nargs=-1, type=click.File('rb'))
+def profile_install_silent(lockdown: LockdownClient, profiles, keystore, keystore_password):
+    """ install given profiles without user interaction (requires the device to be supervised) """
+    service = MobileConfigService(lockdown=lockdown)
+    for profile in profiles:
+        logger.info(f'installing {profile.name}')
+        service.install_profile_silent(
+            profile.read(), keystore.read(), keystore_password)
+
+
 @profile_group.command('cloud-configuration', cls=Command)
 @click.option('--color/--no-color', default=True)
 def profile_cloud_configuration(lockdown: LockdownClient, color):
     """ get cloud configuration """
-    print_json(MobileConfigService(lockdown=lockdown).get_cloud_configuration(), colored=color)
+    print_json(MobileConfigService(
+        lockdown=lockdown).get_cloud_configuration(), colored=color)
 
 
 @profile_group.command('store', cls=Command)

pymobiledevice3/services/mobile_config.py

@@ -5,6 +5,10 @@
 from pymobiledevice3.exceptions import ProfileError
 from pymobiledevice3.lockdown import LockdownClient
 from pymobiledevice3.services.base_service import BaseService
+from cryptography.hazmat.primitives.serialization.pkcs12 import load_pkcs12
+from cryptography.hazmat.primitives.serialization import Encoding
+from cryptography.hazmat.primitives.serialization.pkcs7 import PKCS7SignatureBuilder
+from cryptography.hazmat.primitives import hashes
 
 
 class Purpose(Enum):
@@ -23,23 +27,40 @@ def hello(self) -> None:
     def flush(self) -> None:
         self._send_recv({'RequestType': 'Flush'})
 
+    def escalate(self, pkcs12: bytes, password: str) -> None:
+        decrypted_p12 = load_pkcs12(pkcs12, password.encode('utf-8'))
+
+        escalate_response = self._send_recv({
+            'RequestType': 'Escalate',
+            'SupervisorCertificate': decrypted_p12.cert.certificate.public_bytes(Encoding.DER)
+        })
+        signed_challenge = PKCS7SignatureBuilder().set_data(escalate_response['Challenge']).add_signer(
+            decrypted_p12.cert.certificate, decrypted_p12.key, hashes.SHA256()).sign(Encoding.DER, [])
+        self._send_recv({'RequestType': 'EscalateResponse',
+                        'SignedRequest': signed_challenge})
+        self._send_recv({'RequestType': 'ProceedWithKeybagMigration'})
+
     def get_stored_profile(self, purpose: Purpose = Purpose.PostSetupInstallation) -> Mapping:
         return self._send_recv({'RequestType': 'GetStoredProfile', 'Purpose': purpose.value})
 
     def store_profile(self, profile_data: bytes, purpose: Purpose = Purpose.PostSetupInstallation) -> None:
-        self._send_recv({'RequestType': 'StoreProfile', 'ProfileData': profile_data, 'Purpose': purpose.value})
+        self._send_recv({'RequestType': 'StoreProfile',
+                        'ProfileData': profile_data, 'Purpose': purpose.value})
 
     def get_cloud_configuration(self) -> Mapping:
         return self._send_recv({'RequestType': 'GetCloudConfiguration'}).get('CloudConfiguration')
 
     def set_cloud_configuration(self, cloud_configuration: Mapping) -> None:
-        self._send_recv({'RequestType': 'SetCloudConfiguration', 'CloudConfiguration': cloud_configuration})
+        self._send_recv({'RequestType': 'SetCloudConfiguration',
+                        'CloudConfiguration': cloud_configuration})
 
     def establish_provisional_enrollment(self, nonce: bytes) -> None:
-        self._send_recv({'RequestType': 'EstablishProvisionalEnrollment', 'Nonce': nonce})
+        self._send_recv(
+            {'RequestType': 'EstablishProvisionalEnrollment', 'Nonce': nonce})
 
     def set_wifi_power_state(self, state: bool) -> None:
-        self._send_recv({'RequestType': 'SetWiFiPowerState', 'PowerState': state})
+        self._send_recv(
+            {'RequestType': 'SetWiFiPowerState', 'PowerState': state})
 
     def erase_device(self, preserve_data_plan: bool, disallow_proximity_setup: bool) -> None:
         try:
@@ -54,20 +75,27 @@ def get_profile_list(self) -> Mapping:
     def install_profile(self, payload: bytes) -> None:
         self._send_recv({'RequestType': 'InstallProfile', 'Payload': payload})
 
+    def install_profile_silent(self, profile: bytes, pkcs12: bytes, password: str) -> None:
+        self.escalate(pkcs12, password)
+        self._send_recv(
+            {'RequestType': 'InstallProfileSilent', 'Payload': profile})
+
     def remove_profile(self, identifier: str) -> None:
         profiles = self.get_profile_list()
         if not profiles:
             return
         if identifier not in profiles['ProfileMetadata']:
-            self.logger.info(f'Trying to remove not installed profile: {identifier}')
+            self.logger.info(
+                f'Trying to remove not installed profile: {identifier}')
             return
         meta = profiles['ProfileMetadata'][identifier]
         data = plistlib.dumps({'PayloadType': 'Configuration',
                                'PayloadIdentifier': identifier,
                                'PayloadUUID': meta['PayloadUUID'],
                                'PayloadVersion': meta['PayloadVersion']
                                })
-        self._send_recv({'RequestType': 'RemoveProfile', 'ProfileIdentifier': data})
+        self._send_recv({'RequestType': 'RemoveProfile',
+                        'ProfileIdentifier': data})
 
     def _send_recv(self, request: Mapping) -> Mapping:
         response = self.service.send_recv_plist(request)