#!/usr/bin/env python ############################################################################# # # # Copyright 2020, Digi International Inc. # # # # Permission to use, copy, modify, and/or distribute this software for any # # purpose with or without fee is hereby granted, provided that the above # # copyright notice and this permission notice appear in all copies. # # # # THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES # # WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF # # MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR # # ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES # # WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN # # ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF # # OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE. # # # ############################################################################# from datetime import datetime from digidevice import cli, datapoint from logging.handlers import RotatingFileHandler import argparse import json import logging import re import sys import time import urllib.request import os NAME = "Digi LM940 Modem Firmware Update Script (XOS Version)" VERSION = '1.0.9' LOG_FORMAT = '%(asctime)s %(name)-8s %(levelname)-8s: %(message)s' CONSOLE_FORMAT = '%(asctime)s %(levelname)-8s: %(message)s' MAX_ATTEMPTS = 3 RECHECK_CONNECTION = 60 # seconds BASE_PATH = "http://ftp1.digi.com/support/firmware/transport/LM940_carrier_firmware" LM940_ALL_BIN = "all.bin" LM940_CARRIER = "lm940_carriers.txt" class ModemUpdateException(Exception): pass class MyLogger(object): def __init__(self, name='root', debug=False): self.logger = logging.getLogger() keep_log_count = 2 if debug else 1 fh = RotatingFileHandler(name, maxBytes=100000, backupCount=keep_log_count) ch = logging.StreamHandler() ch.setLevel(logging.DEBUG) self.logger.setLevel(logging.INFO) fh.setLevel(logging.INFO) if debug: self.logger.setLevel(logging.DEBUG) fh.setLevel(logging.DEBUG) fh_format = logging.Formatter(LOG_FORMAT) fh.setFormatter(fh_format) ch_format = logging.Formatter(CONSOLE_FORMAT) ch.setFormatter(ch_format) self.logger.addHandler(fh) self.logger.addHandler(ch) if debug: # immediately rollover logs for ease fh.doRollover() def error(self, msg): self.send_datapoint('ERROR: {0}'.format(msg)) self.logger.error(msg) def warning(self, msg): self.send_datapoint('WARNING: {0}'.format(msg)) self.logger.warning(msg) def info(self, msg): self.send_datapoint(msg) self.logger.info(msg) def debug(self, msg): self.logger.debug(msg) def send_datapoint(self, msg, stream_id="lm940_update"): try: result = datapoint.upload(stream_id, msg) except Exception as err: self.logger.error('Datapoint "{0}/{1}" upload failed: {2}'.format(stream_id, msg, err)) def get_python_autostart_slot(filename): config = cli.execute('show config').splitlines() config_line = None for line in config: if filename in line: config_line = line break if config_line: return config_line.split(' ')[1] return config_line def clear_python_autostart(slot): if slot: cli.execute("python-autostart {slot} args !".format(slot=slot), timeout=10) cli.execute("python-autostart {slot} filepath !".format(slot=slot), timeout=10) cli.execute("python-autostart {slot} on-exit none".format(slot=slot), timeout=10) cli.execute("python-autostart {slot} state on".format(slot=slot), timeout=10) cli.execute("save config", timeout=10) LOGGER.info("clear python-autostart {0} and saved configuration".format(slot)) return LOGGER.info('No python-autostart configured') def get_modem_count(): modems = json.loads(cli.execute('show cellular -fjson')) if len(modems) == 2: LOGGER.debug('get_modem_count: 2') return 2 LOGGER.debug('get_modem_count: 1') return 1 def get_modem_versions(modem_num=1): result = cli.execute('update module {0} show'.format(modem_num)) pattern = re.compile('24\.01\.[0-9]{3}') results = re.findall(pattern, result.strip()) LOGGER.debug('get_modem_versions results: {0}'.format(results)) return results def modem_versions_current(firmware_versions, current_version=['4']): for version in firmware_versions: if version[-1] not in current_version: LOGGER.debug('modem_versions_current: False') return False LOGGER.info('Modem Firmware Current, skipping update, to force update, use `--force_update`') return True def on_wifi(): routes_json = json.loads(cli.execute('show route -fjson')) route_ids = sorted(routes_json) default_route = routes_json['0'] if default_route['Status'].upper() == 'UP' and 'wifi' in default_route['Interface'].lower(): return True return False def on_ethernet(): routes_json = json.loads(cli.execute('show route -fjson')) route_ids = sorted(routes_json) default_route = routes_json['0'] if default_route['Status'].upper() == 'UP' and 'eth' in default_route['Interface'].lower(): return True return False def download_modem_firmware(path, read_bytes=131072): LOGGER.info('Downloading modem firmware file: {0}'.format(path)) start = time.time() LOGGER.debug('... Starting firmware download, {0}'.format(datetime.fromtimestamp(start))) filename = path.split('/')[-1] response = urllib.request.urlopen(path) remaining_bytes = response.length try: with open('./{0}'.format(filename), 'wb') as f: while remaining_bytes > 0: f.write(response.read(read_bytes)) remaining_bytes -= read_bytes time.sleep(0.01) end = time.time() LOGGER.debug('... Finished firmware download, {0}'.format(datetime.fromtimestamp(end))) LOGGER.debug('... Elapsed Seconds: {0}'.format(round(end - start, 2))) except MemoryError: response.close() raise def update_modems(count, filepath_carrier="all"): """ ./ for local file all to pull from Digi FTP """ msg = 'Starting Modem Update, count: {0}'.format(count, filepath_carrier) if filepath_carrier == './': msg += ", file path: {0}".format(filepath_carrier) else: msg += ", carrier: {0}".format(filepath_carrier) LOGGER.info(msg) for i in range(1, count + 1): cmd = "update module {0} {1}".format(i, filepath_carrier) LOGGER.debug('Running cmd: {0}'.format(cmd)) update_module_resp = cli.execute(cmd, timeout=3600) success_msg1 = "Update module {0} succeeded".format(i) success_msg2 = "Module firmware is up-to-date" if success_msg1 not in update_module_resp and success_msg2 not in update_module_resp: LOGGER.debug('Module update result: {0}'.format(update_module_resp)) raise ModemUpdateException('Firmware update on module {0} Failed: {1}'.format(cmd, update_module_resp)) LOGGER.info('Completed update on module: {0}'.format(i)) if count > 1: # Just a safe check between multiple module updates. time.sleep(30) return True def file_present(filename): dir_resp = cli.execute("dir") LOGGER.debug('Checking for {0} in local filesystem'.format(filename)) if filename in dir_resp: return True return False def firmware_paths(basepath, file_list=[LM940_CARRIER, LM940_ALL_BIN]): paths = [] for file in file_list: paths.append("{basepath}/{file}".format(basepath=basepath, file=file)) return paths def remove_update_script(filename): if os.path.isfile(filename): cli.execute('del {0}'.format(filename)) LOGGER.info('Removed {0} from filesystem'.format(filename)) else: LOGGER.warning('{0} is not a file or is not present'.format(filename)) def cloud_connected(): try: status = json.loads(cli.execute('show cloud -fjson'))['status'] if status: return True return False except: return False if __name__ == '__main__': parser = argparse.ArgumentParser( description='{name}, version {ver}'.format(name=NAME, ver=VERSION), formatter_class=argparse.ArgumentDefaultsHelpFormatter) parser.add_argument('--force_update', '-u', action='store_true', help='force Modem Update, even if running current version') parser.add_argument('--cellular_ok', '-c', action='store_true', help='allow download on Cellular connection, will disable connection during update') parser.add_argument('--firmware_path', '-f', default=BASE_PATH, help='web location of firmware file, carrier file must be present in same location') parser.add_argument('--debug', '-d', action='store_true', help='show debug level logging') parser.add_argument('--version', action='version', version=VERSION) args = parser.parse_args() LOGGER = MyLogger('{0}.log'.format(sys.argv[0].split('.')[0]), args.debug) while not cloud_connected(): LOGGER.debug('Not connected to cloud, retry in 10 seconds') time.sleep(10) LOGGER.info('Starting {name}, version {ver}'.format(name=NAME, ver=VERSION)) if args.debug: LOGGER.debug('Running args: {}'.format(args)) RECHECK_CONNECTION = 10 # check for number of modems modem_count = get_modem_count() LOGGER.info('Found {0} modem(s) on router'.format(modem_count)) # check current version for each modem do_update_modem = False current = False if args.force_update: do_update_modem = True else: for i in range(modem_count): module = i + 1 LOGGER.info('Getting Modem Versions for module: {0}'.format(module)) firmware_versions = get_modem_versions(module) current = modem_versions_current(firmware_versions, current_version=['4']) if current: LOGGER.info('Modem Version current on module: {0}'.format(module)) else: do_update_modem = True break # check for Wifi connection success = False attempts = 0 while do_update_modem: if args.cellular_ok or on_wifi() or on_ethernet(): LOGGER.debug('Starting Firmware Update Checks, {0}|{1}'.format(do_update_modem, args.cellular_ok)) LOGGER.info('Checking for firmware files') for firmware_path in firmware_paths(args.firmware_path): filename = firmware_path.split('/')[-1] if not file_present(filename): download_modem_firmware(firmware_path) while attempts < MAX_ATTEMPTS: try: LOGGER.info('Attempt {0} to update modem(s)'.format(attempts + 1)) success = update_modems(modem_count, './') if success: break except ModemUpdateException as err: LOGGER.error(err) attempts += 1 # print('--------------------------------') else: LOGGER.debug('Not on Wifi or Ethernet, waiting {0} seconds for recheck'.format(RECHECK_CONNECTION)) time.sleep(RECHECK_CONNECTION) continue if success: break if attempts >= MAX_ATTEMPTS: LOGGER.error('Too many attempts! Run with `--debug` for more information') break # print('===================================') if success or current: autostart_slot = get_python_autostart_slot(sys.argv[0]) if autostart_slot: clear_python_autostart(autostart_slot) if not args.debug: remove_update_script(sys.argv[0]) time.sleep(0.5) # make sure this log is sent as a datapoint LOGGER.info('Update Complete for Modem Firmware') else: LOGGER.error('Update NOT Successful for Modem Firmware')