#!/usr/bin/python # software_update.py # --copyright-- Copyright 2011 (C) Tranzoa, Co. All rights reserved. Warranty: You're free and on your own here. This code is not necessarily up-to-date or of public quality. # --url-- http://www.tranzoa.net/tzpython/ # --email-- pycode is the name to send to. tranzoa.com is the place to send to. # --bodstamps-- # November 5, 2011 bar # November 6, 2011 bar keep doing failure detection badly ( .failed ) # March 22, 2012 bar allow caller to spec the version regx string # March 26, 2012 bar another log msg # April 19, 2012 bar signature file crc xxxxXXXX tzlib.blkcrc32 of sorted (case-sensitive path/name) (file name + file content) crcs # --eodstamps-- ## \file # # # Do a software update from a zip file. # # This code finds the latest zip file in the given directory named: BASIC_FILE_NAME_V_MM_mm.zip, # where MM is the major version and mm is the minor. # Then this code unzips the zip file to .tmp files in the given directory (which is made if it doesn't exist), # validating that there is a file, version.py, .txt or .htm, containing a version number matching the zip file name. # Then this code renames the .tmp files, chmod's any .py and .sh and un-ext files to executable. # The code checks for a file in the zip file called SCRIPT and, if so, passes its name back to the caller whether there was a failure or not. # # import glob import os import re import stat import zipfile import tzlib import replace_file BASIC_FILE_NAME = "update" # the zip file name without _V_MM_mm.zip VERSION_FILE_NAME = "version.htm" # a file that must be in the zip file. It has the version in it. For double-checking FORCE_FILE = "force_update.txt" # if this file is in the zip file, then the update is done no matter whether the version is new or not SCRIPT_FILE_NAME = "update.py" # if this file is in the zip file, its name is returned to the caller SIGNATURE_FILE_NAME = "update_signature.txt" # if this file is in the zip file and the caller sets the signature file name (to this name or any other), then this file's contents are used to validate the rest of the zip file contents EXIT_DELAY = 15.0 crc_sig_re = re.compile(r"^crc\s+([0-9a-f]{8})\s*$", re.MULTILINE + re.IGNORECASE) class a_software_updater(object) : def log(me, s) : # callback stub pass def doing_update(me, doing_it) : # callback stub return(True) # allow update to happen def fail(me, s) : # callback stub pass def _failed(me, s) : me.failed = True me.on_fail(s) def __init__(me, current_version = None, zip_file_basic_name = None, version_file_name = None, script_file_name = None, signature_file_name = None, to_dir = None, v_regx_str = None, on_log = None, on_doing_update = None, on_fail = None, ) : me.current_version = current_version or "" me.zip_file_basic_name = zip_file_basic_name or BASIC_FILE_NAME me.version_file_name = version_file_name or VERSION_FILE_NAME me.script_file_name = script_file_name or SCRIPT_FILE_NAME me.signature_file_name = signature_file_name or None me.v_regx_str = v_regx_str or r"\d+\.\d+" me.to_dir = to_dir or "" me.on_log = on_log or me.log me.on_doing_update = on_doing_update or me.doing_update me.on_fail = on_fail or me.fail me.update_zfn = "" # most recent zip file found me.failed = False def do_update(me, zip_dr, to_dir = None, current_version = None, zip_file_basic_name = None, version_file_name = None, script_file_name = None, signature_file_name = None, ) : to_dir = to_dir or me.to_dir or "." current_version = current_version or me.current_version or "" me.zip_file_basic_name = zip_file_basic_name or me.zip_file_basic_name or BASIC_FILE_NAME me.version_file_name = version_file_name or me.version_file_name or VERSION_FILE_NAME me.script_file_name = script_file_name or me.script_file_name or SCRIPT_FILE_NAME me.signature_file_name = signature_file_name or me.signature_file_name or None script = "" me.failed = False ( zfn, ver ) = me.find_zip_file(zip_dr, current_version) if (me.update_zfn != zfn) and zfn : # is there a new update zip file - one we've not already done? me.update_zfn = zfn # don't do the same file again fns = me.unzip_zip_file(zfn, ver, to_dir) if len(fns) and me.on_doing_update(True) : # if the zip file unzipped OK and all was well with it, let's rename over the new files and set their permissions and etc. me.on_log("Software update: %u files unzipped." % len(fns)) script = me.rename_updated_files(fns, to_dir) me.on_doing_update(False) pass return(script) # return the update script file name if there was one whether the update fully succeeded or not def find_zip_file(me, zip_dr, current_version) : zfn = "" ver = current_version ufn = os.path.splitext(os.path.basename(me.zip_file_basic_name))[0] fns = glob.glob(os.path.join(zip_dr, ufn + "*.zip")) for fn in fns : g = re.search(r"(?i)_V_(" + me.v_regx_str.replace('.', '_') + ")\.zip$", fn) if g : v = g.group(1).replace("_", ".") if ver < v : # find the highest version update zip file in the directory ver = v zfn = fn pass pass return(zfn, ver) def unzip_zip_file(me, zfn, ver, to_dir) : ok = False fns = [] try : me.on_log("Reading software update from [%s]" % zfn ) zf = zipfile.ZipFile(zfn, "r") fns = zf.namelist() if (me.version_file_name not in fns) and (FORCE_FILE not in fns) : s = "Software update: without %s or %s in [%s]" % ( me.version_file_name, FORCE_FILE, zfn ) me.on_log(s) me._failed(s) else : ok = True if (not FORCE_FILE in fns) and (me.version_file_name in fns) : # check version number if not forced fi = zf.open(me.version_file_name, "r") fd = fi.read() fi.close() g = re.search(r'(?ms)^VERSION\s+\=\s+"(' + me.v_regx_str + ')"', fd) if not g : g = re.search(r'(' + me.v_regx_str + ')', fd) if not g : s = "Softare update: VERSION not found in %s:%s" % ( zfn, me.version_file_name ) me.on_log(s) me._failed(s) ok = False else : sver = g.group(1) if sver != ver : s = "Softare update: %s VERSION %s different from file name: %s" % ( me.version_file_name, sver, zfn ) me.on_log(s) me._failed(s) ok = False pass pass if ok : crc_exp = None crc_sigs = {} for fn in fns : fi = zf.open(fn, "r") fd = fi.read() fi.close() if fn == me.signature_file_name : g = crc_sig_re.search(fd) if g : crc_exp = int(g.group(1), 16) pass else : crc_sigs[fn] = tzlib.blkcrc32(tzlib.blkcrc32(tzlib.INITIAL_CRC32_VALUE, fn), fd) if fn != FORCE_FILE : me.on_log("Updating %s %u %08x" % ( fn, len(fd), crc_sigs[fn] ) ) try : dn = os.path.join(to_dir, os.path.dirname(fn)) if dn and (not os.path.exists(dn)) and (not os.path.isdir(dn)) : os.makedirs(dn) tzlib.write_whole_binary_file(os.path.join(to_dir, fn + ".tmp"), fd) except ( IOError, OSError ) : s = "Software update: failure writing [%s]" % fn me.on_log(s) me._failed(s) ok = False break pass pass pass if crc_exp != None : kys = crc_sigs.keys() kys.sort() crc = tzlib.INITIAL_CRC32_VALUE for fn in kys : crc = tzlib.blkcrc32(crc, "%08x" % crc_sigs[fn]) # the CRC is the crc of the crcs of each of the file name+contents, sorted by file name, case-sensitive, including directory names if crc != crc_exp : ok = False s = "ZIP file CRC mismatch %08x != exptected %08x [%s]!" % ( crc, crc_exp, zfn ) me.on_log(s) me._failed(s) pass if me.signature_file_name : try : fns.remove(me.signature_file_name) # don't try to rename, etc, this file - because we did not write it except ValueError : pass pass pass pass zf.close() except ( OSError, IOError ) : s = "Bad ZIP file or file read error [%s]!" % zfn me.on_log(s) me._failed(s) ok = False try : zf.close() except ( OSError, IOError ) : pass pass return((ok and fns) or []) def rename_updated_files(me, fns, to_dir) : script = "" ok = True for fn in fns : if fn != FORCE_FILE : fn = os.path.join(to_dir, fn) try : bfn = fn + ".bak" tfn = fn + ".tmp" if os.path.isfile(fn) : mode = os.stat(fn).st_mode # get the mode of the old version of the file else : mode = os.stat(tfn).st_mode | stat.S_IXUSR | stat.S_IXGRP replace_file.replace_file(fn, fn + ".tmp", bfn) # rename the .tmp file to the proper name if os.path.basename(fn) == me.script_file_name : script = os.path.abspath(fn) # remember that the zip file had an update script file, the name of which we return to the caller if (fn.endswith(".sh") or fn.endswith(".py") or (not os.path.splitext(fn)[1])) and os.path.isfile(fn) : mode |= stat.S_IXUSR | stat.S_IXGRP # no matter what, let's just make .sh and .py files executable try : os.chmod(fn, mode) except ( IOError, OSError ) : ok = False s = "Software update: failure chmod'ing [%s]" % fn me.on_log(s) me._failed(s) pass if fn.endswith(".py") : cfn = os.path.splitext(fn)[0] + ".pyc" if os.path.isfile(cfn) : try : os.remove(cfn) except ( IOError, OSError ) : ok = False s = "Software update: failure to remove [%s]" % cfn me.on_log(s) me._failed(s) pass pass pass except ( IOError, OSError ) : ok = False s = "Software update: failure renaming or stat'ing [%s]" % fn me.on_log(s) me._failed(s) # !!!! hard FAULT forever pass pass if ok : me.on_log("Update script: [%s]" % script) return(script) # a_software_updater if __name__ == '__main__' : def log(s) : print s def doing_update(h) : print "DOING", h return(True) def fail(s) : print "FAIL:", s me = a_software_updater(to_dir = "utest", on_log = log, on_doing_update = doing_update, on_fail = fail) print me.do_update(".", to_dir = "utest_it") # # # eof