diff --git a/app_vsens.zip b/app_vsens.zip new file mode 100644 index 0000000..8974bc8 Binary files /dev/null and b/app_vsens.zip differ diff --git a/application.json b/application.json new file mode 100644 index 0000000..e69de29 diff --git a/ble_legacy_dfu_controller.py b/ble_legacy_dfu_controller.py index e315c10..edfc403 100644 --- a/ble_legacy_dfu_controller.py +++ b/ble_legacy_dfu_controller.py @@ -197,7 +197,7 @@ def check_DFU_mode(self): try: res = self.ble_conn.expect('handle:.*', timeout=10) # res = self.ble_conn.expect('handle:', timeout=10) - except pexpect.TIMEOUT, e: + except pexpect.TIMEOUT as e: print("State timeout") except: pass diff --git a/ble_secure_dfu_controller.py b/ble_secure_dfu_controller.py index e2aed62..8eeb5fa 100644 --- a/ble_secure_dfu_controller.py +++ b/ble_secure_dfu_controller.py @@ -1,6 +1,7 @@ import math import pexpect import time +import binascii from array import array from util import * @@ -71,9 +72,9 @@ def from_string(res_str): class BleDfuControllerSecure(NrfBleDfuController): # Class constants - UUID_BUTTONLESS = '8e400001-f315-4f60-9fb8-838830daea50' UUID_CONTROL_POINT = '8ec90001-f315-4f60-9fb8-838830daea50' UUID_PACKET = '8ec90002-f315-4f60-9fb8-838830daea50' + UUID_BUTTONLESS = '8ec90003-f315-4f60-9fb8-838830daea50' # Constructor inherited from abstract base class @@ -83,7 +84,6 @@ class BleDfuControllerSecure(NrfBleDfuController): def start(self): (_, self.ctrlpt_handle, self.ctrlpt_cccd_handle) = self._get_handles(self.UUID_CONTROL_POINT) (_, self.data_handle, _) = self._get_handles(self.UUID_PACKET) - if verbose: print('Control Point Handle: 0x%04x, CCCD: 0x%04x' % (self.ctrlpt_handle, self.ctrlpt_cccd_handle)) print('Packet handle: 0x%04x' % (self.data_handle)) @@ -111,8 +111,8 @@ def check_DFU_mode(self): dfu_mode = False try: - self.ble_conn.expect([self.UUID_BUTTONLESS], timeout=2) - except (pexpect.TIMEOUT , e): + self.ble_conn.expect([self.UUID_BUTTONLESS], timeout=5) + except pexpect.TIMEOUT as e: dfu_mode = True return dfu_mode @@ -120,14 +120,15 @@ def check_DFU_mode(self): def switch_to_dfu_mode(self): (_, bl_value_handle, bl_cccd_handle) = self._get_handles(self.UUID_BUTTONLESS) - self._enable_notifications(bl_cccd_handle) + self._enable_indications(bl_cccd_handle) # Reset the board in DFU mode. After reset the board will be disconnected cmd = 'char-write-req 0x%04x 01' % (bl_value_handle) self.ble_conn.sendline(cmd) # Wait some time for board to reboot - time.sleep(0.5) + time.sleep(1) + print("Switched to DFU mode Successfully") # Increase the mac address by one and reconnect self.target_mac_increase(1) @@ -189,7 +190,7 @@ def _wait_and_parse_notify(self): if result[1] != Results.SUCCESS: raise Exception("Error in {} procedure, reason: {}".format( Procedures.to_string(result[0]), - Results.to_string(result[1]))) + Results.to_string(result[1]))) return result @@ -274,14 +275,14 @@ def _dfu_send_object(self, offset, obj_max_size): if offset == 0 or offset >= obj_max_size or crc32 != crc32_unsigned(self.bin_array[0:offset]): # Create Data Object size = min(obj_max_size, self.image_size - offset) - self._dfu_send_command(Procedures.CREATE, [Procedures.PARAM_DATA] + uint32_to_bytes_le(size)) + self._dfu_send_command(Procedures.CREATE, [Procedures.PARAM_DATA] + uint32_to_bytes_le(int(size))) self._wait_and_parse_notify() segment_count = 0 segment_total = int(math.ceil(min(obj_max_size, self.image_size-offset)/float(self.pkt_payload_size))) - segment_begin = offset - segment_end = min(offset+obj_max_size, self.image_size) + segment_begin = int(offset) + segment_end = int(min(offset+obj_max_size, self.image_size)) for i in range(segment_begin, segment_end, self.pkt_payload_size): num_bytes = min(self.pkt_payload_size, segment_end - i) @@ -301,7 +302,7 @@ def _dfu_send_object(self, offset, obj_max_size): if res != Results.SUCCESS: raise Exception("bad notification status: {}".format(Results.to_string(res))) - + if crc32 != crc32_unsigned(self.bin_array[0:offset]): # Something went wrong, need to re-transmit this object return 0 diff --git a/dfu.py b/dfu.py index b0be5dd..49aa1f0 100755 --- a/dfu.py +++ b/dfu.py @@ -19,7 +19,7 @@ def main(): - init_msg = """ + init_msg = """ ================================ == == == DFU Server == @@ -81,7 +81,7 @@ def main(): options, args = parser.parse_args() - except (Exception , e): + except Exception as e: print(e) print("For help use --help") sys.exit(2) @@ -108,7 +108,7 @@ def main(): #print(options.zipfile) try: hexfile, datfile = unpacker.unpack_zipfile(options.zipfile) - except (Exception, e): + except Exception as e: print("ERR") print(e) pass @@ -137,6 +137,7 @@ def main(): else: ble_dfu = BleDfuControllerLegacy(options.address.upper(), hexfile, datfile) + # Initialize inputs ble_dfu.input_setup() @@ -147,21 +148,26 @@ def main(): success = ble_dfu.switch_to_dfu_mode() if not success: print("Couldn't reconnect") + else: + ble_dfu.start() + else: + print("Already in DFU mode") + ble_dfu.start() else: + print("Couldn't connect, will try DFU MAC") # The device might already be in DFU mode (MAC + 1) ble_dfu.target_mac_increase(1) # Try connection with new address - print("Couldn't connect, will try DFU MAC") - if not ble_dfu.scan_and_connect(): + if ble_dfu.scan_and_connect(): + ble_dfu.start() + else: raise Exception("Can't connect to device") - ble_dfu.start() - # Disconnect from peer device if not done already and clean up. ble_dfu.disconnect() - except (Exception, e): + except Exception as e: # print(traceback.format_exc()) print("Exception at line {}: {}".format(sys.exc_info()[2].tb_lineno, e)) pass @@ -171,7 +177,7 @@ def main(): # If Unpacker for zipfile used then delete Unpacker if unpacker != None: - unpacker.delete() + unpacker.delete() print("DFU Server done") diff --git a/manifest.json b/manifest.json new file mode 100644 index 0000000..c8723a8 --- /dev/null +++ b/manifest.json @@ -0,0 +1,8 @@ +{ + "manifest": { + "application": { + "bin_file": "ble_app_vsens_pca10100_s113.bin", + "dat_file": "ble_app_vsens_pca10100_s113.dat" + } + } +} \ No newline at end of file diff --git a/nrf_ble_dfu_controller.py b/nrf_ble_dfu_controller.py index b1b2671..f5a8829 100644 --- a/nrf_ble_dfu_controller.py +++ b/nrf_ble_dfu_controller.py @@ -1,10 +1,12 @@ import os import pexpect import re +import zlib from abc import ABCMeta, abstractmethod from array import array from util import * +from scan import Scan verbose = False @@ -60,9 +62,7 @@ def __init__(self, target_mac, firmware_path, datfile_path): self.firmware_path = firmware_path self.datfile_path = datfile_path - - self.ble_conn = pexpect.spawn("sudo hcitool lescan") - self.ble_conn = pexpect.spawn("gatttool -b '%s' --interactive" % target_mac) + self.ble_conn = pexpect.spawn("gatttool -b '%s' -t random --interactive" % target_mac) self.ble_conn.delaybeforesend = 0 # -------------------------------------------------------------------------- @@ -105,7 +105,6 @@ def input_setup(self): self.image_size = len(self.bin_array) print("Binary imge size: %d" % self.image_size) - print("Binary CRC32: %d" % crc32_unsigned(array_to_hex_string(self.bin_array))) return @@ -122,21 +121,26 @@ def input_setup(self): # Perform a scan and connect via gatttool. # Will return True if a connection was established, False otherwise # -------------------------------------------------------------------------- - def scan_and_connect(self, timeout=2): + def scan_and_connect(self, timeout=30): if verbose: print("scan_and_connect") + self.scan_obj = Scan(None) + scan_list = self.scan_obj.scan() + print("scan_list:",scan_list) print("Connecting to %s" % (self.target_mac)) try: self.ble_conn.expect('\[LE\]>', timeout=timeout) - except (pexpect.TIMEOUT, e): + except pexpect.TIMEOUT as e: + print("scan error:",e) return False self.ble_conn.sendline('connect') try: res = self.ble_conn.expect('.*Connection successful.*', timeout=timeout) - except (pexpect.TIMEOUT, e): + print("Connection successfull") + except pexpect.TIMEOUT as e: return False return True @@ -153,8 +157,7 @@ def target_mac_increase(self, inc): # Re-start gatttool with the new address self.disconnect() - self.ble_conn = pexpect.spawn("sudo hcitool lescan") - self.ble_conn = pexpect.spawn("gatttool -b '%s' --interactive" % self.target_mac) + self.ble_conn = pexpect.spawn("gatttool -b '%s' -t random --interactive" % self.target_mac) self.ble_conn.delaybeforesend = 0 # -------------------------------------------------------------------------- @@ -167,10 +170,11 @@ def _get_handles(self, uuid): self.ble_conn.sendline('characteristics') try: - self.ble_conn.expect([uuid], timeout=2) - handles = re.findall('.*handle: (0x....),.*char value handle: (0x....)', self.ble_conn.before) + self.ble_conn.expect([uuid], timeout=10) + handles = re.findall('.*handle: (0x....),.*char value handle: (0x....)', self.ble_conn.before.decode('utf-8')) (handle, value_handle) = handles[-1] - except (pexpect.TIMEOUT, e): + + except pexpect.TIMEOUT as e: raise Exception("UUID not found: {}".format(uuid)) return (int(handle, 16), int(value_handle, 16), int(value_handle, 16)+1) @@ -202,7 +206,7 @@ def _dfu_wait_for_notify(self): # continue to wait. # self.ble_conn.sendline('') - string = self.ble_conn.before + string = self.ble_conn.before.decode('utf-8') if '[ ]' in string: print('Connection lost! ') raise Exception('Connection Lost') @@ -211,7 +215,7 @@ def _dfu_wait_for_notify(self): if index == 0: after = self.ble_conn.after hxstr = after.split()[3:] - handle = long(float.fromhex(hxstr[0])) + handle = int(float.fromhex(hxstr[0].decode('utf-8'))) return hxstr[2:] else: @@ -234,7 +238,8 @@ def _dfu_send_command(self, procedure, params=[]): # Verify that command was successfully written try: res = self.ble_conn.expect('Characteristic value was written successfully.*', timeout=10) - except (pexpect.TIMEOUT, e): + except pexpect.TIMEOUT as e: + print("State timeout") # -------------------------------------------------------------------------- @@ -264,5 +269,23 @@ def _enable_notifications(self, cccd_handle): # Verify that command was successfully written try: res = self.ble_conn.expect('Characteristic value was written successfully.*', timeout=10) - except (pexpect.TIMEOUT, e): + except pexpect.TIMEOUT as e: + print("State timeout") + + # -------------------------------------------------------------------------- + # Enable notifications from the Control Point Handle + # -------------------------------------------------------------------------- + def _enable_indications(self, cccd_handle): + if verbose: print('_enable_notifications') + + cmd = 'char-write-req 0x%04x %s' % (cccd_handle, '0200') + + if verbose: print(cmd) + + self.ble_conn.sendline(cmd) + + # Verify that command was successfully written + try: + res = self.ble_conn.expect('Characteristic value was written successfully.*', timeout=10) + except pexpect.TIMEOUT as e: print("State timeout") diff --git a/scan.py b/scan.py index 12f576b..9e67b7a 100755 --- a/scan.py +++ b/scan.py @@ -8,6 +8,8 @@ import pexpect import signal +import sys +import time #------------------------------------------------------------------------------ # Bluetooth LE scan for advertising peripheral devices @@ -21,10 +23,16 @@ def __init__( self, advert_name ): def scan( self ): try: - self.hcitool = pexpect.spawn('hcitool lescan') + self.hcitool = pexpect.spawn('hciconfig hci0 down') + self.hcitool = pexpect.spawn('hciconfig hci0 up') + self.hcitool = pexpect.spawn('hcitool lescan') #self.hcitool.logfile = sys.stdout - index = self.hcitool.expect(['LE Scan ...'], 1) + index = self.hcitool.expect(['LE Scan ...']) + time.sleep(10) + print("scan after:",self.hcitool.after) + except pexpect.EOF: + self.hcitool.terminate(force=True) return [] except Exception as err: @@ -38,10 +46,13 @@ def scan( self ): return [] list = [] - for dummy in range(0, 2): + for dummy in range(0, 4): + #print("for loop:",self.hcitool.readline()) try: list.append(self.hcitool.readline()) + #print("list:",list) except pexpect.TIMEOUT: + return list break if list == []: @@ -60,7 +71,7 @@ def scan( self ): # Strip newline from items in list list = [item.strip() for item in list] - list = list[0:2] + #list = list # Close pexpect (release device for subsequent use) self.hcitool.terminate(force=True) @@ -91,6 +102,7 @@ def scan(self): print("scan: pexpect.TIMEOUT") pass except Exception as e: + print(e) print("scan: exception: {0} ".format(sys.exc_info()[0])) pass diff --git a/unpacker.py b/unpacker.py index 1df03fd..f0a34df 100755 --- a/unpacker.py +++ b/unpacker.py @@ -15,7 +15,7 @@ class Unpacker(object): # #-------------------------------------------------------------------------- def entropy(self, length): - return ''.join(random.choice(string.lowercase) for i in range (length)) + return ''.join(random.choice(string.ascii_lowercase) for i in range (length)) #-------------------------------------------------------------------------- # @@ -26,7 +26,7 @@ def unpack_zipfile(self, file): raise Exception("Error: file, not found!") # Create unique working direction into which the zip file is expanded - self.unzip_dir = "{0}/{1}_{2}".format(tempfile.gettempdir(), os.path.splitext(basename(file))[0], self.entropy(6)) + self.unzip_dir = ("{0}/{1}_{2}".format(tempfile.gettempdir(), os.path.splitext(basename(file))[0], self.entropy(6))) datfilename = "" binfilename = "" diff --git a/util.py b/util.py index b67da7f..34fd669 100644 --- a/util.py +++ b/util.py @@ -29,15 +29,14 @@ def array_to_hex_string(arr): return hex_str def crc32_unsigned(bytestring): - return binascii.crc32(bytestring) % (1 << 32) + return binascii.crc32(bytestring) % (1 << 32) def mac_string_to_uint(mac): parts = list(re.match('(..):(..):(..):(..):(..):(..)', mac).groups()) - ints = map(lambda x: int(x, 16), parts) - + ints = list(map(lambda x: int(x, 16), parts)) res = 0 - for i in range(0, len(ints)): - res += (ints[len(ints)-1 - i] << 8*i) + for i in range(0, len((ints))): + res += (ints[len((ints))-1 - i] << 8*i) return res