feat: Add option to query installed app path

Author: Berstanio

src/main/java/io/github/berstanio/pymobiledevice3/ipc/PyMobileDevice3IPC.java

@@ -209,6 +209,44 @@ public CompletableFuture<DeviceInfo> getDevice(String uuid) {
         });
     }
 
+    /**
+     * @param appPath The path to the app bundle, to retrieve the identifier from
+     * @return The bundle identifier. Fails exceptionally, if no bundle identifier can be found
+     */
+    public CompletableFuture<String> getBundleIdentifier(File appPath)
+    {
+        JSONObject object = new JSONObject();
+        object.put("command", "get_bundle_identifier");
+        object.put("app_path", appPath.getAbsolutePath());
+
+        return createRequest(object, (future, jsonObject) -> {
+            future.complete(jsonObject.getString("result"));
+        });
+    }
+
+    /**
+     * Gets the installed path of an app on a specific device
+     *
+     * @param info The device to query
+     * @param bundleIdentifier The bundle identifier to search for
+     * @return The on-device path or null, if the bundle is not installed
+     */
+    public CompletableFuture<String> getInstalledPath(DeviceInfo info, String bundleIdentifier)
+    {
+        JSONObject object = new JSONObject();
+        object.put("command", "get_installed_path");
+        object.put("device_id", info.getUniqueDeviceId());
+        object.put("bundle_identifier", bundleIdentifier);
+
+        return createRequest(object, (future, jsonObject) -> {
+            if (jsonObject.get("state").equals("not_installed")) {
+                future.complete(null);
+            } else {
+                future.complete(jsonObject.getString("result"));
+            }
+        });
+    }
+
     /**
      * @param deviceInfo The device to install to.
      * @param path .app bundle path

src/main/resources/handler.py

@@ -13,6 +13,7 @@
 
 from threading import Thread
 from time import sleep
+from typing import Union
 
 import requests
 from packaging.version import Version
@@ -165,7 +166,7 @@ def get_device(id, device_id, ipc_client):
         with create_using_usbmux(device_id, autopair=False) as lockdown:
             reply = {"id": id, "state": "completed", "result": lockdown.short_info}
             write_dispatcher.write_reply(ipc_client, reply)
-    except NoDeviceConnectedError | DeviceNotFoundError:
+    except Union[NoDeviceConnectedError, DeviceNotFoundError]:
         reply = {"id": id, "state": "failed_expected"}
         write_dispatcher.write_reply(ipc_client, reply)
 
@@ -199,6 +200,27 @@ def progress_handler(progress, *args):
 
         print("Installed bundle: " + str(bundle_identifier))
 
+def get_bundle_identifier(id, path, ipc_client):
+    info_plist_path = Path(path) / "Info.plist"
+    with open(info_plist_path, 'rb') as f:
+        plist_data = plistlib.load(f)
+
+    bundle_identifier = plist_data["CFBundleIdentifier"]
+
+    reply = {"id": id, "state": "completed", "result": bundle_identifier}
+    write_dispatcher.write_reply(ipc_client, reply)
+
+def get_installed_path(id, lockdown_client, bundle_identifier, ipc_client):
+    with InstallationProxyService(lockdown=lockdown_client) as installer:
+        res = installer.lookup(options={"BundleIDs" : [bundle_identifier]})
+
+        if bundle_identifier not in res:
+            reply = {"id": id, "state": "not_installed"}
+        else:
+            reply = {"id": id, "state": "completed", "result": res[bundle_identifier]["Path"]}
+
+        write_dispatcher.write_reply(ipc_client, reply)
+
 def decode_plist(id, path, ipc_client):
     with open(path, 'rb') as f:
         plist_data = plistlib.load(f)
@@ -402,6 +424,9 @@ def handle_command(command, ipc_client):
         elif command_type == "get_version":
             get_version(id, ipc_client)
             return
+        elif command_type == "get_bundle_identifier":
+            get_bundle_identifier(id, res["app_path"], ipc_client)
+            return
 
         # Now come the device targetted functions
         device_id = res['device_id']
@@ -416,6 +441,9 @@ def handle_command(command, ipc_client):
                 port = res['port'] if 'port' in res else 0
                 debugserver_connect(id, lockdown, port, ipc_client)
                 return
+            elif command_type == "get_installed_path":
+                get_installed_path(id, lockdown, res["bundle_identifier"], ipc_client)
+                return
 
     except Exception as e:
         reply = {"request": command, "state": "failed", "error": repr(e), "backtrace": traceback.format_exc()}