update-ledger.py

#! /usr/bin/env python3
#
# Depends on btchip-python and websockets_client
# 
# Do:
# pip install btchip-python
# pip install websocket_client
#

import argparse
import asyncio
import binascii
import atexit
import json
import requests
import struct
import time

from btchip.btchipComm import getDongle
from btchip.btchipException import BTChipException
from urllib.parse import urlencode
from websocket import create_connection

PROVIDERS = {
  '': 1,
  'das': 2,
  'club': 3,
  'shitcoins': 4,
  'ee': 5
}

def get_device_info(dongle):
    out = {
        'target_id': 0,
        'se_version': '',
        'mcu_version': '',
        'provider': 1,
        'version': ''
    }

    data = dongle.exchange(bytearray([0xe0, 0x01, 0x00, 0x00, 0x00]))
    pos = 0

    target_id_bytes = data[0:4]
    out['target_id'] = struct.unpack('>I', target_id_bytes)[0]
    pos += 4

    se_version_len = data[pos]
    pos += 1
    se_version_bytes = data[pos:pos + se_version_len]
    out['se_version'] = se_version_bytes.decode()
    out['version'] = out['se_version'].replace('-osu', '')
    pos += se_version_len

    flags_len = data[pos]
    pos += 1
    flags = data[pos:pos + flags_len]
    pos += flags_len

    if pos >= len(data):
        return out

    mcu_version_len = data[pos]
    pos += 1
    mcu_version_bytes = data[pos:pos + mcu_version_len]

    if (mcu_version_bytes[-1] == 0):
        mcu_version_bytes = mcu_version_bytes[:-1]
    out['mcu_version'] = mcu_version_bytes.decode()

    provider_split = out['se_version'].split('-')
    provider_name = ''
    if len(provider_split) > 1:
        provider_name = provider_split[1]
    if provider_name in PROVIDERS:
        out['provider'] = PROVIDERS[provider_name]

    return out

def install_via_websocket(url, dongle):
    ws = create_connection(url)
    while True:
        # Get data from server
        server_msg = ws.recv()
        data = json.loads(server_msg)
        if args.debug:
            print(data)
        if data['query'] == 'success':
            break
        if data['query'] == 'error':
            ws.close()
            print('Error: {}'.format(data))
            exit(-1)

        # Send to device and build response
        response = {
            'nonce': data['nonce']
        }
        apdus = []
        try:
            if data['query'] == 'exchange':
                apdus.append(binascii.unhexlify(data['data']))
            elif data['query'] == 'bulk':
                for apdu in data['data']:
                    apdus.append(binascii.unhexlify(apdu))
            else:
                response['response'] = 'unsupported'
        except:
            reponse['reponse'] = 'parse error'

        if len(apdus) > 0:
            try:
                for apdu in apdus:
                    response['data'] = binascii.hexlify(dongle.exchange(apdu)).decode()
                if len(response['data']) == 0:
                    response['data'] = "9000"
                response['response'] = "success"
            except Exception as e:
                if args.debug:
                    print(str(e))
                response['response'] = "I/O"

        # Reply to the server
        if args.debug:
            print(response)
        ws.send(json.dumps(response))
    ws.close()

def wait_for_dongle(bootloader_mode):
    while True:
        try:
            print('.', end='')
            dongle = getDongle(args.debug)
            dev_info = get_device_info(dongle)
            # Check if it is in bootloader mode
            if bootloader_mode:
                if (dev_info['target_id'] & 0xf0000000) != 0x30000000:
                    break
            else:
                break
            dongle.close()
        except Exception as e:
            if args.debug:
                import traceback
                traceback.print_exc()
            time.sleep(1)
    return dongle, dev_info

if __name__ == '__main__':

    parser = argparse.ArgumentParser(description='Updates the firmware on a Ledger Nano S')
    parser.add_argument('--firmware', help='The firmware version to install. Default is the latest')
    parser.add_argument('--debug', help='Enable debug logging', action='store_true')
    parser.add_argument('--stage', help='Which stage of update to perform', choices=['all', 'osu', 'mcu', 'final'], default='all')
    args = parser.parse_args()

    dongle = getDongle(args.debug)

    # Cleanup
    def cleanup_dongle():
        dongle.close()
    atexit.register(cleanup_dongle)

    # Get some device information
    device_info = get_device_info(dongle)

    is_bootloader = (device_info['target_id'] & 0xf0000000) != 0x30000000

    # Get MCU information
    mcu_id = -1
    if not is_bootloader:
        mcu_resp = requests.get('https://manager.api.live.ledger.com/api/mcu_versions')
        mcus = mcu_resp.json()
        print(device_info)
        for mcu in mcus:
            if device_info['mcu_version'] == mcu['name']:
                mcu_id = mcu['id']
                break
        else:
            print('Did not find MCU')
            exit(-1)

        # Get device version information
        data = {'provider': device_info['provider'], 'target_id': device_info['target_id']}
        device_ver_resp = requests.post('https://manager.api.live.ledger.com/api/get_device_version', data=data)
        device_ver = device_ver_resp.json()

        device_ver_id = device_ver['id']

    
    if args.stage == 'all' or args.stage == 'osu':
        # Get information about the firmware
        data = {'provider': device_info['provider'], 'version_name': device_info['version'], 'device_version': device_ver_id}
        firmware_ver_resp = requests.post('https://manager.api.live.ledger.com/api/get_firmware_version', data=data)
        firmware_ver = firmware_ver_resp.json()

        firmware_ver_id = firmware_ver['id']

        # When firmware version is specified
        if args.firmware is not None:
            # Get all of the firmware versions
            all_firm_resp = requests.get('https://manager.api.live.ledger.com/api/firmware_final_versions')
            all_firm = all_firm_resp.json()

            for firm in all_firm:
                if args.firmware == firm['name']:
                    for osu in firm['osu_versions']:
                        if firmware_ver_id in osu['previous_se_firmware_final_versions'] and device_ver_id in osu['device_versions']:
                            latest_osu = osu
                            break
                    else:
                        print('Cannot upgrade device to this firmware')
                        exit(-1)
                    break
            else:
                print('Did not find specified firmware')
                exit(-1)
        else:
            # Get information about OSU for latest firmware
            data = {'current_se_firmware_final_version': firmware_ver_id, 'device_version': device_ver_id, 'provider': device_info['provider']}
            latest_osu_resp = requests.post('https://manager.api.live.ledger.com/api/get_latest_firmware?livecommonversion=deadbeef', data=data)
            latest_osu_json = latest_osu_resp.json()
            if latest_osu_json['result'] == 'null':
                print("Already up to date")
                exit(1)
            else:
                latest_osu = latest_osu_json['se_firmware_osu_version']

        new_firmware_id = latest_osu['next_se_firmware_final_version']
        firmware_osu = latest_osu['firmware']
        firmware_key_osu = latest_osu['firmware_key']
        perso_osu = latest_osu['perso']

        # Get info about latest firmware
        latest_firm_resp = requests.get('https://manager.api.live.ledger.com/api/firmware_final_versions/{}'.format(new_firmware_id))
        latest_firm = latest_firm_resp.json()

        mcu_versions = latest_firm['mcu_versions']
        firmware = latest_firm['firmware']
        firmware_key = latest_firm['firmware_key']
        perso = latest_firm['perso']

        # Install OSU firmware
        osu_url_params = {
            'targetId': device_info['target_id'],
            'firmware': firmware_osu,
            'firmwareKey': firmware_key_osu,
            'perso': perso_osu
        }
        print(osu_url_params)
        osu_install_url = 'wss://api.ledgerwallet.com/update/install?{}'.format(urlencode(osu_url_params))
        install_via_websocket(osu_install_url, dongle)

    if args.stage == 'final' or args.stage == 'mcu':
        if args.firmware is not None:
            # Get all of the firmware versions
            all_firm_resp = requests.get('https://manager.api.live.ledger.com/api/firmware_final_versions')
            all_firm = all_firm_resp.json()

            for firm in all_firm:
                if args.firmware == firm['name']:
                    firmware = firm['firmware']
                    firmware_key = firm['firmware_key']
                    perso = firm['perso']
                    mcu_versions = firm['mcu_versions']
                    print(mcu_versions)
                    break
            else:
                print('Did not find specified firmware')
                exit(-1)
        else:
            print('Firmware version must be specified for mcu and final stages')
            exit(-1)

    if (args.stage == 'all' or args.stage == 'mcu') and mcu_id not in mcu_versions:
        # Get new MCU info
        new_mcu_id = mcu_versions[0]
        new_mcu_resp = requests.get('https://manager.api.live.ledger.com/api/mcu_versions/{}'.format(new_mcu_id))
        new_mcu = new_mcu_resp.json()

        new_mcu_name = new_mcu['name']

        # Prompt to replug in bootloader mode
        print('Please unplug your device.')
        print('Then plug it back in while holding both buttons in order to boot into bootloader mode')
        dongle.close()
        dongle, device_info = wait_for_dongle(True)

        # Install the MCU
        mcu_url_params = {
            'targetId': device_info['target_id'],
            'version': new_mcu_name
        }
        mcu_install_url = 'wss://api.ledgerwallet.com/update/mcu?{}'.format(urlencode(mcu_url_params))
        install_via_websocket(mcu_install_url, dongle)
        dongle.close()
        dongle, device_info = wait_for_dongle(False)

    if args.stage == 'all' or args.stage == 'final':
        # Install final firmware
        firmware_url_params = {
            'targetId': device_info['target_id'],
            'firmware': firmware,
            'firmwareKey': firmware_key,
            'perso': perso
        }
        print(firmware_url_params)
        firmware_url = 'wss://api.ledgerwallet.com/update/install?{}'.format(urlencode(firmware_url_params))
        install_via_websocket(firmware_url, dongle)

For anyone interested in this, it's fairly straightforward to update this script to get it to work:

• Pass a correct ledgercommonversion to the first three calls (for instance 31.0.0)
• The POST payload should be a JSON string, so update the first three calls from data=data to data=json.dumps(data)
• The websocket API url has changed