iOS app installation

server/mobile.py

@@ -6,10 +6,12 @@
 import subprocess
 import base64
 import json
-from typing import Literal
+from typing import Literal, Optional
 import asyncio
 import socket
 import xml.etree.ElementTree as ET
+import zipfile
+import plistlib
 
 import requests
 from androguard.core.apk import APK
@@ -253,7 +255,7 @@
 
 def run_adb_command(command):
     """Run an ADB command and return the output."""
     try:
         result = subprocess.run(
             command,
             shell=True,
@@ -404,9 +406,6 @@
             return
     except:
         pass
-
-    # No real source available
-    raise Exception("iOS service error. Make sure simulator is running.")
 
 
 async def upload_android_ui_dump():
@@ -533,3 +532,187 @@
     except Exception as e:
         return {"message": f"Error installing APK: {str(e)}", "filename": filename, "package_name": package_name}
 
+
+@router.get("/package-installed")
+def check_package_installed(package_name: str, serial: str | None = None):
+    """Check if an android package is installed on the device."""
+    try:
+        device_flag = f"-s {serial}" if serial else ""
+        result = run_adb_command(
+            f"{ADB_PATH} {device_flag} shell pm list packages {package_name}".strip()
+        )
+        if result.startswith("Error:"):
+            return {"installed": False, "package_name": package_name, "error": result}
+
+        # pm list packages returns lines like "package:com.example.app"
+        # Check if the exact package is in the output
+        installed_packages = [line.replace("package:", "") for line in result.split("\n") if line.startswith("package:")]
+        is_installed = package_name in installed_packages
+
+        return {"installed": is_installed, "package_name": package_name}
+    except Exception as e:
+        return {"installed": False, "package_name": package_name, "error": str(e)}
+
+
+@router.post("/ios/app-upload")
+def handle_ios_app_upload(file: UploadFile = File(...)):
+    dir_path = f"{ZEUZ_NODE_DOWNLOADS_DIR}/ios-app"
+    if not os.path.exists(dir_path):
+        os.makedirs(dir_path)
+
+    filename = file.filename or "uploaded.app"
+    filepath = os.path.join(dir_path, filename)
+    with open(filepath, "wb") as buffer:
+        shutil.copyfileobj(file.file, buffer)
+
+    return {"message": "App uploaded successfully", "filename": filename}
+
+
+def normalized_ios_app_path(file_path: str) -> Optional[str]:
+    """
+    ensure that we ended up with a .app directory even if user provided .ipa
+    """
+    if not os.path.exists(file_path):
+        return None
+
+    # if already .app
+    if file_path.endswith(".app") and os.path.isdir(file_path):
+        return file_path
+
+    # .ipa, so we have to extract to get .app
+    if file_path.endswith(".ipa"):
+        extract_dir = file_path.replace(".ipa", "_extracted")
+        zip_path = file_path.replace(".ipa", ".zip")
+
+        # copy instead of rename to avoid breaking original file
+        shutil.copy(file_path, zip_path)
+
+        with zipfile.ZipFile(zip_path, "r") as zip_ref:
+            zip_ref.extractall(extract_dir)
+
+        payload_dir = os.path.join(extract_dir, "Payload")
+        if not os.path.isdir(payload_dir):
+            return None
+
+        for item in os.listdir(payload_dir):
+            if item.endswith(".app"):
+                return os.path.join(payload_dir, item)
+
+        return None
+
+    # extract zip because 'test.app' is not a file, so browser zip that before send to me
+    if file_path.endswith(".app.zip") or file_path.endswith(".zip"):
+        extract_dir = file_path + "_extracted"
+        os.makedirs(extract_dir, exist_ok=True)
+
+        with zipfile.ZipFile(file_path, "r") as zip_ref:
+            zip_ref.extractall(extract_dir)
+
+        # search recursively for .app folder so it will pick very first .app folder
+        for root, dirs, files in os.walk(extract_dir):
+            for d in dirs:
+                if d.endswith(".app") and os.path.isdir(os.path.join(root, d)):
+                    return os.path.join(root, d)
+
+        return None
+
+    return None
+
+
+def extract_bundle_id_from_app(app_path: str) -> Optional[str]:
+    """
+    Reads CFBundleIdentifier from Info.plist inside .app bundle
+    """
+    info_plist_path = os.path.join(app_path, "Info.plist")
+
+    if not os.path.exists(info_plist_path):
+        return None
+
+    try:
+        with open(info_plist_path, "rb") as f:
+            plist_data = plistlib.load(f)
+            return plist_data.get("CFBundleIdentifier")
+    except Exception:
+        return None
+
+
+@router.post("/ios/app-install")
+def handle_ios_app_install(filename: str, sim_udid: str):
+    """
+    handling the ios app installation in the simolator
+    """
+    dirpath = f"{ZEUZ_NODE_DOWNLOADS_DIR}/ios-app"
+    filepath = os.path.join(dirpath, filename)
+    if not os.path.exists(filepath):
+        return {"message": "App not found", "filename": filename}
+
+    app_path = normalized_ios_app_path(filepath)
+    if not app_path:
+        return {"message": "Failed to normalize .app", "filename": filename}
+
+    bundle_id = extract_bundle_id_from_app(app_path)
+
+    try:
+        # uninstall if already exists
+        try:
+            subprocess.run(
+                ["xcrun", "simctl", "uninstall", sim_udid, bundle_id],
+                capture_output=True, text=True, timeout=30
+            )
+        except:
+            pass  # Ignore uninstall errors
+
+        # Install the app
+        result = subprocess.run(
+            ["xcrun", "simctl", "install", sim_udid, app_path],
+            capture_output=True, text=True, check=True, timeout=120
+        )
+
+        # Verify installation
+        verify_result = subprocess.run(
+            ["xcrun", "simctl", "get_app_container", sim_udid, bundle_id],
+            capture_output=True, text=True, timeout=30
+        )
+
+        if verify_result.returncode != 0:
+            return {
+                "message": f"App installed but verification failed: {verify_result.stderr}",
+                "filename": filename,
+                "bundle_id": bundle_id,
+            }
+
+        return {
+            "message": "App installed successfully",
+            "filename": filename,
+            "bundle_id": bundle_id,
+        }
+    except subprocess.CalledProcessError as e:
+        return {
+            "message": f"Error installing app: {e.stderr or str(e)}",
+            "filename": filename,
+            "bundle_id": bundle_id,
+        }
+
+
+@router.get("/ios/bundle-installed")
+def is_ios_app_installed(sim_udid: str, bundle_id: str):
+    try:
+        result = subprocess.run(
+            ["xcrun", "simctl", "get_app_container", sim_udid, bundle_id],
+            capture_output=True, text=True, timeout=10
+        )
+
+        if result.returncode == 0 and result.stdout.strip():
+            list_result = subprocess.run(
+                ["xcrun", "simctl", "listapps", sim_udid],
+                capture_output=True, text=True, timeout=10
+            )
+
+            if list_result.returncode == 0:
+                return {"installed": bundle_id in list_result.stdout}
+
+            return {"installed": True}
+
+        return {"installed": False}
+    except Exception as e:
+        return {"installed": False, "error": str(e)}