Example Script

This example shell script demonstrates how to use the Corellium Python API to:

1. Search for a device with a specified instance ID.
2. If it's found, ensure the device is powered on.
3. If it's not found, create a new device with a desired hardware and firmware and wait for it to be ready.
4. Wait for the device to boot.
5. Wait for the device's app subsystem to be ready.
6. List the running apps on the device.

Pre-Requisites

1. Ensure the following pip dependencies are installed on your local environment:

   • corellium_api

2. Set an environmental variable CORELLIUM_API_TOKEN with your API token.

3. Set another environmental variable CORELLIUM_API_ENDPOINT.

   • For On-Premise customers, use https://<your_hostname>/api, where <your_hostname> is either the hostname or IP address of your Corellium appliance.
   • For Enterprise Cloud customers, use https://<your_domain>.enterprise.corellium.com/api, where <your_domain> is taken from the URL you use to access the Corellium web interface.
   • For Solo Cloud customers, use https://app.corellium.com/api.

Example Script

import os
import sys
import time
import asyncio
from asyncio import TimeoutError
import corellium_api
from corellium_api.rest import ApiException

# Define the default values for the instance
DEFAULT_INSTANCE_ID = 'YOUR_INSTANCE_IDENTIFIER'
DEFAULT_INSTANCE_FIRMWARE = '18.0'
DEFAULT_INSTANCE_HARDWARE = 'iphone16pm'
DEFAULT_INSTANCE_NAME = 'Python_API_demo_device'
DEFAULT_PROJECT_ID = 'YOUR_PROJECT_IDENTIFIER'

# Define time and retry constants
SLEEP_TIME_SECONDS = 60
MAX_RETRIES = 10
WAIT_TIMEOUT_SECONDS = 300

# Define constants for instance states and task states
INSTANCE_STATE_ON = 'on'
INSTANCE_STATE_OFF = 'off'
INSTANCE_STATE_BOOTING = 'booting'
INSTANCE_STATE_CREATING = 'creating'
INSTANCE_STATE_REBOOTING = 'rebooting'
INSTANCE_TASK_STATE_NONE = 'none'

required_env_vars = ['CORELLIUM_API_ENDPOINT', 'CORELLIUM_API_TOKEN']
for var in required_env_vars:
    if not os.getenv(var):
        print(f'Error: Environment variable {var} is not set.', file=sys.stderr)
        sys.exit(1)

CORELLIUM_API_ENDPOINT = os.getenv('CORELLIUM_API_ENDPOINT')
CORELLIUM_API_TOKEN = os.getenv('CORELLIUM_API_TOKEN')

configuration = corellium_api.Configuration(host=CORELLIUM_API_ENDPOINT)
configuration.access_token = CORELLIUM_API_TOKEN

def handle_unexpected_state(instance_id, instance_state, instance_task_state):
    raise RuntimeError(f'Unexpected state, instance {instance_id} is {instance_state} with taskState {instance_task_state}.')

async def print_list_of_running_apps(api_instance, instance_id):
    try:
        api_response_list_apps = await api_instance.v1_list_apps(instance_id)
        print(f'List of running apps on instance {instance_id}:')
        for app in api_response_list_apps:
            print(f'  {app.bundleId} ({app.name})')
    except ApiException as e:
        raise RuntimeError(f'Exception when calling CorelliumApi->v1_list_apps: {e}\n')
    except Exception as e:
        raise RuntimeError(f'Exception in print_list_of_running_apps: {e}\n')

async def turn_instance_on_or_create_new_instance(api_instance):
    try:
        api_response_get_default_instance = await api_instance.v1_get_instance(DEFAULT_INSTANCE_ID)
        print(f'Instance {DEFAULT_INSTANCE_ID} was found.')
        # check if the instance is off
        if api_response_get_default_instance.state == INSTANCE_STATE_OFF:
            print(f'Instance {DEFAULT_INSTANCE_ID} is off.')
            # start the instance
            await api_instance.v1_start_instance(DEFAULT_INSTANCE_ID)
            print(f'Instance {DEFAULT_INSTANCE_ID} is powering on.')
        return DEFAULT_INSTANCE_ID
    except ApiException as e:
        if e.status == 404:
            print(f'Instance {DEFAULT_INSTANCE_ID} was not found.')
            print(f'Creating a new instance of {DEFAULT_INSTANCE_HARDWARE} hardware with {DEFAULT_INSTANCE_FIRMWARE} firmware.')
            try:
                api_response_create_instance = await api_instance.v1_create_instance({
                    "project": DEFAULT_PROJECT_ID,
                    "name": DEFAULT_INSTANCE_NAME,
                    "flavor": DEFAULT_INSTANCE_HARDWARE,
                    "os": DEFAULT_INSTANCE_FIRMWARE
                })
                print(f'Instance {api_response_create_instance.id} is creating.')
                return api_response_create_instance.id
            except ApiException as e:
                if e.status == 400 and 'insufficient resource: cores' in e.body:
                    raise RuntimeError(f'Insufficient resources to create a new instance: {e.body}')
                else:
                    raise RuntimeError(f'Exception when calling CorelliumApi->v1_create_instance: {e}\n')
            except Exception as e:
                raise RuntimeError(f'Exception in turn_instance_on_or_create_new_instance: {e}\n')
        else:
            raise RuntimeError(f'Exception when calling CorelliumApi->v1_get_instance: {e}\n')
    except Exception as e:
        raise RuntimeError(f'Exception in turn_instance_on_or_create_new_instance: {e}\n')

async def wait_for_instance_agent_to_be_ready(api_instance, instance_id):
    retries = 0
    while retries <= MAX_RETRIES:
        if retries == MAX_RETRIES:
            raise RuntimeError(f'Failed to check if the app subsystem is ready after {MAX_RETRIES} retries.')
        try:
            print('Checking if the app subsystem is ready.')
            api_response_agent_status = await api_instance.v1_agent_app_ready(instance_id)
            if api_response_agent_status.ready:
                print(f'The app subsystem on instance {instance_id} is ready.')
                return
        except ApiException as e:
            if e.status == 504:
                retries += 1
                await asyncio.sleep(SLEEP_TIME_SECONDS)
            else:
                raise RuntimeError(f'Exception when calling CorelliumApi->v1_agent_app_ready: {e}\n')
        except Exception as e:
            raise RuntimeError(f'Exception in wait_for_instance_agent_to_be_ready: {e}\n')

async def wait_for_instance_to_turn_on(api_instance, instance_id):
    instance_state = None
    instance_task_state = None
    while instance_state != INSTANCE_STATE_ON or instance_task_state != INSTANCE_TASK_STATE_NONE:
        try:
            api_response_get_instance = await api_instance.v1_get_instance(instance_id)
            instance_state = api_response_get_instance.state
            instance_task_state = api_response_get_instance.task_state
            if instance_state == INSTANCE_STATE_ON and instance_task_state == INSTANCE_TASK_STATE_NONE:
                print(f'Instance {instance_id} is on.')
                break
            elif instance_state in [INSTANCE_STATE_BOOTING, INSTANCE_STATE_CREATING, INSTANCE_STATE_REBOOTING]:
                print(f'Waiting for instance to finish {instance_state}.')
            else:
                handle_unexpected_state(instance_id, instance_state, instance_task_state)
            await asyncio.sleep(SLEEP_TIME_SECONDS)
        except ApiException as e:
            raise RuntimeError(f"Exception when calling CorelliumApi->v1_get_instance: {e}\n")
        except Exception as e:
            raise RuntimeError(f"Exception in wait_for_instance_to_turn_on: {e}\n")

async def main():
    print(f'Starting the script at {time.strftime("%Y-%m-%d %H:%M:%S")}.')
    new_instance_id = None
    async with corellium_api.ApiClient(configuration) as api_client:
        try:
            api_instance = corellium_api.CorelliumApi(api_client)
            instance_id = await asyncio.wait_for(turn_instance_on_or_create_new_instance(api_instance), timeout=WAIT_TIMEOUT_SECONDS)
            await asyncio.wait_for(wait_for_instance_to_turn_on(api_instance, instance_id), timeout=WAIT_TIMEOUT_SECONDS)
            await asyncio.wait_for(wait_for_instance_agent_to_be_ready(api_instance, instance_id), timeout=WAIT_TIMEOUT_SECONDS)
            await asyncio.wait_for(print_list_of_running_apps(api_instance, instance_id), timeout=WAIT_TIMEOUT_SECONDS)
#        except TimeoutError as e:
#            print(f'Operation timed out after 1 hour: {e}', file=sys.stderr)
#            sys.exit(1)
        except RuntimeError as e:
            print(f'RuntimeError: {e}', file=sys.stderr)
            sys.exit(1)

if __name__ == '__main__':
    asyncio.run(main())