aioservices/aiomqtt: Fix firmware match re &

* Fix pattern matching with '*' wildcards and

* Allow asyncmd ota server serve local firmware files (filesystem path as
name) as well as remote files (URLs paths as names) e.g.:
- /path/to/firmware/build-ESP32_OTA/micropython.bin
- http://myserver.local:8000/firmware/build-ESP32_OTA/micropython.bin

Author: Carglglz

aioservices/services/aiomqtt_service.py

@@ -12,6 +12,7 @@
 import io
 import aiostats
 import machine
+import re
 
 
 class MQTTService(Service):
@@ -107,6 +108,20 @@ def _taskinfo(self):
         self._services_total = len(aioctl.tasks_match("*.service"))
         self._ctasks_total = len(aioctl.tasks_match("*.service.*"))
 
+    def _grep(self, patt, filename):
+        if isinstance(patt, list):
+            pass
+        else:
+            patt = [patt]
+        _pattlst = (
+            re.compile(_patt.replace(".", r"\.").replace("*", ".*") + "$")
+            for _patt in patt
+        )
+        try:
+            return any(_pattrn.match(filename) for _pattrn in _pattlst)
+        except Exception:
+            return None
+
     @aioctl.aiotask
     async def do_action(self, action, service):
         if action == "status":
@@ -329,7 +344,7 @@ async def do_action(self, action, service):
                                 if self.log:
                                     self.log.info("No new OTA update")
                                 return
-                            elif not _ota_params["fwfile"].endswith(self._fwfile):
+                            elif not self._grep(self._fwfile, _ota_params["fwfile"]):
                                 if self.log:
                                     self.log.info("No new OTA update")
                                 return
@@ -347,7 +362,9 @@ async def do_action(self, action, service):
                                 {
                                     "notify": False,
                                     "sha": _ota_service._comp_sha_ota("", rtn=True),
-                                    "fwfile": self._fwfile,
+                                    "fwfile": self._fwfile
+                                    if not self._fwf_re
+                                    else _ota_params["fwfile"],
                                     "ip": (
                                         aioctl.group()
                                         .tasks["network.service"]

cli/asyncmd/async_ota.py

@@ -12,6 +12,10 @@
 import json
 import asyncio
 import time
+from urllib.parse import urlparse
+import urllib.request
+import aiohttp
+import tempfile
 
 BLOCKLEN = 4096
 
@@ -31,6 +35,14 @@
     pb = False
 
 
+def is_url(url):
+    try:
+        result = urlparse(url)
+        return all([result.scheme, result.netloc])
+    except ValueError:
+        return False
+
+
 def do_pg_bar(
     fwrite,
     index,
@@ -165,14 +177,56 @@ def __init__(
         self.log.info("Serving firmware:")
         for fwfile in firmwares:
             self.log.info(f"- {fwfile}")
-            if os.path.exists(fwfile):
-                self.update_sha(fwfile)
+            self.update_sha(fwfile)
 
     def update_sha(self, fwfile):
-        # self._fw_file = firmware
-        with open(fwfile, "rb") as fwr:
-            firmware = fwr.read()
+        if os.path.exists(fwfile):
+            with open(fwfile, "rb") as fwr:
+                firmware = fwr.read()
+        elif is_url(fwfile):
+            try:
+                firmware = self.http_get(fwfile)
+            except Exception:
+                self.log.error(f"Firmware: {fwfile} not available")
+                return
+        else:
+            self.log.error(f"Firmware: {fwfile} not available")
+
+            return
+
+        self._fw_files[fwfile]["sz"] = sz = len(firmware)
+        if is_url(fwfile):
+            self._fw_files[fwfile]["data"] = firmware
+        hf = hashlib.sha256(firmware)
+        if sz % BLOCKLEN != 0:
+            self._fw_files[fwfile]["n_blocks"] = _n_blocks = (sz // BLOCKLEN) + 1
+
+            hf.update(b"\xff" * ((_n_blocks * BLOCKLEN) - sz))
+        else:
+            self._fw_files[fwfile]["n_blocks"] = _n_blocks = sz // BLOCKLEN
+        self.check_sha[fwfile] = self._fw_files[fwfile]["sha"] = hexlify(
+            hf.digest()
+        ).decode()
+
+    async def async_update_sha(self, fwfile):
+        if os.path.exists(fwfile):
+            with open(fwfile, "rb") as fwr:
+                firmware = fwr.read()
+        elif is_url(fwfile):
+            try:
+                firmware = await self.aiohttp_get(fwfile)
+            except Exception:
+                self.log.error(f"Firmware: {fwfile} not available")
+                return
+        else:
+            self.log.error(f"Firmware: {fwfile} not available")
+
+            return
+
         self._fw_files[fwfile]["sz"] = sz = len(firmware)
+
+        if is_url(fwfile):
+            self._fw_files[fwfile]["data"] = firmware
         hf = hashlib.sha256(firmware)
         if sz % BLOCKLEN != 0:
             self._fw_files[fwfile]["n_blocks"] = _n_blocks = (sz // BLOCKLEN) + 1
@@ -184,6 +238,22 @@ def update_sha(self, fwfile):
             hf.digest()
         ).decode()
 
+    def http_get(self, fwfile):
+        resp = urllib.request.urlopen(fwfile)
+        if resp.status == 200:
+            firmware = resp.read()
+            resp.close()
+            return firmware
+        else:
+            raise ValueError
+
+    async def aiohttp_get(self, fwfile):
+        async with aiohttp.ClientSession() as client:
+            async with client.get(fwfile) as resp:
+                assert resp.status == 200
+                firmware = await resp.read()
+                return firmware
+
     def find_localip(self):
         ip_soc = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
         ip_soc.connect(("8.8.8.8", 1))
@@ -363,13 +433,27 @@ async def do_async_ota(self, addr, reader, writer):
 
         sz = self._fw_files[fwfile]["sz"]
         if not self._bg:
-            tqdm.write(f"{os.path.relpath(fwfile)}  [{sz / 1000:.2f} kB]")
+            if not is_url(fwfile):
+                tqdm.write(f"{os.path.relpath(fwfile)}  [{sz / 1000:.2f} kB]")
+            else:
+                tqdm.write(f"{fwfile}  [{sz / 1000:.2f} kB]")
         cnt = 0
         data = await reader.read(2)
         assert data == b"OK"
         _pb_index = self.get_pos_pgb()
         buff = b""
         t_start = time.time()
+        tf = None
+
+        if is_url(fwfile):
+            # url_path = urlparse(fwfile).path
+            url_fwf = fwfile
+            tf = tempfile.NamedTemporaryFile(delete=False)
+            tf.write(self._fw_files[fwfile]["data"])
+            tf.seek(0)
+            fwfile = tf.name
+            tf.close()
+
         with open(fwfile, "rb") as fmwf:
             with cstqdm(
                 desc=_dev,
@@ -417,7 +501,10 @@ async def do_async_ota(self, addr, reader, writer):
                         pass
 
         tqdm.write("")
-        tqdm.write(f"{os.path.basename(fwfile)} @ {_dev} [ \033[92mOK\x1b[0m ] ")
+        if tf is None:
+            tqdm.write(f"{os.path.basename(fwfile)} @ {_dev} [ \033[92mOK\x1b[0m ] ")
+        else:
+            tqdm.write(f"{url_fwf} @ {_dev} [ \033[92mOK\x1b[0m ] ")
         do_pg_bar(
             tqdm.write,
             loop_index,
@@ -432,6 +519,9 @@ async def do_async_ota(self, addr, reader, writer):
         )
         tqdm.write("\n")
         self._busy_pos.remove(_pb_index)
+        # Remove temp file
+        if tf and os.path.exists(tf.name):
+            os.remove(tf.name)
         if self._async:
             while True:
                 data = await reader.read(2)

cli/bin/asyncmd

@@ -20,6 +20,7 @@ from asyncmd.status import get_status
 import yaml
 import requests
 import time
+from urllib.parse import urlparse
 
 _CONFIG_DIR = os.path.join(os.environ["HOME"], ".asyncmd")
 _CONFIG_FILE = os.path.join(_CONFIG_DIR, "asyncmd.config")
@@ -300,6 +301,14 @@ fh_err.setFormatter(fmt_err)
 log.addHandler(fh_err)
 
 
+def is_url(url):
+    try:
+        result = urlparse(url)
+        return all([result.scheme, result.netloc])
+    except ValueError:
+        return False
+
+
 def get_sha256(file):
     try:
         with open(file, "rb") as fwf:
@@ -342,7 +351,9 @@ async def _ota(args, log, tls_params=None, ota_tls_params={}):
 async def async_ota(args, log, tls_params=None, ota_tls_params={}):
     tasks = set()
     interval = 5
+    url_files = [fwfile for fwfile in args.ff if is_url(fwfile)]
     args.ff = [os.path.abspath(fwfile) for fwfile in args.ff if os.path.exists(fwfile)]
+    args.ff += url_files
 
     while True:
         try:
@@ -365,9 +376,13 @@ async def async_ota(args, log, tls_params=None, ota_tls_params={}):
                 otaserv_sha_task = asyncio.create_task(
                     ota_sha_check_listen(args, log, ota_server, tls_params)
                 )
+                check_http_sha_task = asyncio.create_task(
+                    check_http_server_fwfiles(args, log, tls_params, ota_server)
+                )
                 tasks.add(ota_task)
                 tasks.add(check_sha_task)
                 tasks.add(otaserv_sha_task)
+                tasks.add(check_http_sha_task)
                 # add ota_check listener
                 await asyncio.gather(*tasks)
         except aiomqtt.MqttError:
@@ -417,20 +432,42 @@ async def ota_sha_check_listen(args, log, server, tls_params=None):
                     sha_file["fwfile"] = args.ff[0]
 
                 if sha_file["fwfile"] not in server.check_sha:
-                    log.error(
-                        f"OTA request from {devname}: FILE "
-                        f": {sha_file['fwfile']} not available"
-                    )
-                    continue
+                    if not any(
+                        [
+                            fwfile.endswith(sha_file["fwfile"].replace("*", ""))
+                            for fwfile in server.check_sha
+                        ]
+                    ):
+                        log.error(
+                            f"OTA request from {devname}: FILE "
+                            f": {sha_file['fwfile']} not available"
+                        )
+                        continue
 
-                log.info(
-                    f"OTA request from {devname}: FILE: "
-                    f"{os.path.relpath(sha_file['fwfile'])}"
-                )
+                if not is_url(sha_file["fwfile"]) and not sha_file["fwfile"].startswith(
+                    "*"
+                ):
+                    log.info(
+                        f"OTA request from {devname}: FILE: "
+                        f"{os.path.relpath(sha_file['fwfile'])}"
+                    )
+                else:
+                    log.info(
+                        f"OTA request from {devname}: FILE: " f"{sha_file['fwfile']}"
+                    )
                 log.info(f"OTA request from {devname}: SHA: {sha_file['sha']}")
 
                 # get device from topic
                 # publish to device/{name}/ota
+                # match sha_file
+                if sha_file["fwfile"].startswith("*"):
+                    _matchs = [
+                        fwfile
+                        for fwfile in server.check_sha
+                        if fwfile.endswith(sha_file["fwfile"].replace("*", ""))
+                    ]
+                    if _matchs:
+                        sha_file["fwfile"] = _matchs[0]
 
                 server.register_device(devname, sha_file)
 
@@ -501,7 +538,7 @@ async def check_fw_sha(args, log, client, server):
                     log.info("New or modified firmware file detected")
                     log.info(f"{os.path.relpath(fwfile)}: SHA256: {fw_sha[fwfile]}")
                     log.info("Sending notification...")
-                    server.update_sha(fwfile)  # fix this
+                    await server.async_update_sha(fwfile)  # fix this
                     await client.publish(
                         args.t,
                         payload=json.dumps(
@@ -523,6 +560,45 @@ async def check_fw_sha(args, log, client, server):
         await asyncio.sleep(5)
 
 
+async def check_http_server_fwfiles(args, log, tls_params, server):
+    sub_topic = "asyncmd/otacheck"
+
+    fw_sha = {fwfile: "" for fwfile in args.ff if is_url(fwfile)}
+
+    async with aiomqtt.Client(
+        hostname=args.ht, port=args.p, logger=log, tls_params=tls_params
+    ) as client:
+        await client.subscribe(sub_topic)
+        async with client.messages() as messages:
+            async for message in messages:
+                try:
+                    fw_sha_server = json.loads(message.payload.decode())
+                    for fwfile in fw_sha_server:
+                        if fw_sha.get(fwfile) != fw_sha_server.get(fwfile):
+                            fw_sha[fwfile] = fw_sha_server.get(fwfile)
+
+                            log.info("New or modified firmware file detected")
+                            log.info(f"{fwfile}: SHA256: {fw_sha[fwfile]}")
+                            log.info("Sending notification...")
+                            await server.async_update_sha(fwfile)  # fix this
+                            await client.publish(
+                                args.t,
+                                payload=json.dumps(
+                                    {
+                                        "host": server.host,
+                                        "port": server.port,
+                                        "sha": server.check_sha[fwfile],
+                                        "blocks": server._fw_files[fwfile]["n_blocks"],
+                                        "bg": server._bg,
+                                        "fwfile": fwfile,
+                                    }
+                                ),
+                            )
+
+                except Exception as e:
+                    log.error(e)
+
+
 async def check_device_sha(args, log, tls_params):
     pub_topic = f"device/{args.d}/ota"
     _msg = "check"