"""OTA update - filesystem and ZIP helpers.
Module-level functions and constants of ``firmware_core.py`` that
:meth:`findylab.findyIoT.findyIoT.downloadUpdate`,
:meth:`findylab.findyIoT.findyIoT.update` and ``boot.py`` build on. The
package is a ZIP archive that **must be stored uncompressed** - build it
with ``zip -0 -r -X fw.zip .``; DEFLATE entries are only handled when the
device build ships the ``deflate`` module.
Download and apply are two independent, restartable phases and the final
swap onto the live tree is committed at boot from a staging directory,
so a power loss or network drop at any point is safe:
* drop while downloading - resume from the byte offset already on flash;
* drop while extracting - no ``OTA_COMMIT`` marker, the package is
discarded, retry command 52;
* drop while boot-committing - marker still present, the copy re-runs
(idempotent);
* new firmware will not import - ``boot.py`` falls back to ``base.py``.
See :doc:`/guides/ota-update` for the whole flow.
"""
import struct
#: Download chunk size in bytes (LTE MTU-sized; was 512 on 2G).
OTA_CHUNK = 1460
#: Re-entrancy guard for ``downloadUpdate()``.
_ota_busy = False
#: The package being downloaded.
OTA_ZIP = '/update.zip'
#: JSON pinning the partial download to a package name, size and chunk size.
OTA_META = '/update.meta'
#: Staging directory the package is extracted into.
OTA_STAGE = '/next'
#: Commit marker - ``boot.py`` copies ``OTA_STAGE`` onto the live tree while it exists.
OTA_COMMIT = '/update.commit'
#: Written by ``boot.py`` after a successful commit; consumed by the next report.
OTA_APPLIED = '/update.applied'
#: ZIP local file header signature.
_ZIP_LOCAL_SIG = b'PK\x03\x04'
[docs]
def _ota_mkdirs(path):
"""mkdir -p for a '/a/b/c' path (ignores already-existing dirs)."""
import os
cur = ''
for part in path.split('/'):
if not part:
continue
cur += '/' + part
try:
os.mkdir(cur)
except OSError:
pass
[docs]
def _ota_rmtree(path):
"""rm -rf, tolerant of a missing path."""
import os
try:
st = os.stat(path)
except OSError:
return
if st[0] & 0x4000: # directory bit
for ent in os.ilistdir(path):
_ota_rmtree(path + '/' + ent[0])
try:
os.rmdir(path)
except OSError:
pass
else:
try:
os.remove(path)
except OSError:
pass
[docs]
def _zip_entries(f):
"""Yield (name, method, crc, csize, usize, data_pos) for every local
file header in an open zip, seeking f to the next header after each
item. Stops at the central directory. Raises ValueError on a streaming
data descriptor (flag bit 3): `zip -0 -X` never sets it, but a wrongly
built archive would, and we must not guess sizes for an OTA image."""
while True:
sig = f.read(4)
if sig != _ZIP_LOCAL_SIG:
break
hdr = f.read(26)
if len(hdr) < 26:
break
(ver, flag, method, mtime, mdate,
crc, csize, usize, fnlen, extralen) = struct.unpack('<HHHHHIIIHH', hdr)
if flag & 0x08:
raise ValueError("zip has data descriptor - rebuild with zip -0 -X")
name = f.read(fnlen).decode('utf-8')
if extralen:
f.read(extralen)
data_pos = f.tell()
yield (name, method, crc, csize, usize, data_pos)
f.seek(data_pos + csize)
[docs]
def _ota_inflate_crc(f, csize, sink):
"""Inflate a DEFLATE entry (f positioned at its data), pushing plaintext
into sink(bytes) if given and returning its CRC32 - or None if the build
has no `deflate` module."""
try:
import deflate, io, binascii
except ImportError:
return None
raw = f.read(csize)
d = deflate.DeflateIO(io.BytesIO(raw), deflate.RAW)
c = 0
while True:
b = d.read(1024)
if not b:
break
c = binascii.crc32(b, c)
if sink:
sink(b)
return c & 0xffffffff
[docs]
def ota_zip_ok(path):
"""Verify every entry's CRC32 end-to-end. True if the whole archive is
intact, False on any mismatch/parse error, None if it needs DEFLATE
support the build lacks."""
import binascii
try:
f = open(path, 'rb')
except OSError:
return False
try:
for (name, method, crc, csize, usize, data_pos) in _zip_entries(f):
if name.endswith('/'):
continue
f.seek(data_pos)
if method == 0:
c = 0
remaining = csize
while remaining:
b = f.read(min(1024, remaining))
if not b:
return False
c = binascii.crc32(b, c)
remaining -= len(b)
if (c & 0xffffffff) != crc:
print("OTA: CRC mismatch", name)
return False
elif method == 8:
c = _ota_inflate_crc(f, csize, None)
if c is None:
print("OTA: no deflate support for", name)
return None
if c != crc:
print("OTA: CRC mismatch", name)
return False
else:
print("OTA: unsupported method", method, name)
return None
return True
except Exception as e:
print("OTA: verify error", e)
return False
finally:
f.close()