west: blobs: verify fetched blobs after downloading

scripts/west_commands/blobs.py

@@ -119,19 +119,43 @@ def fetch_blob(self, url, path):
         self.ensure_folder(path)
         inst.fetch(url, path)
 
+    # Compare the checksum of a file we've just downloaded
+    # to the digest in blob metadata, warn user if they differ.
+    def verify_blob(self, blob):
+        log.dbg('Verifying blob {module}: {abspath}'.format(**blob))
+
+        status = zephyr_module.get_blob_status(blob['abspath'], blob['sha256'])
+        if status == zephyr_module.BLOB_OUTDATED:
+            log.err(textwrap.dedent(
+                f'''\
+                The checksum of the downloaded file does not match that
+                in the blob metadata:
+                - if it is not certain that the download was successful,
+                  try running 'west blobs fetch {blob['module']}'
+                  to re-download the file
+                - if the error persists, please consider contacting
+                  the maintainers of the module so that they can check
+                  the corresponding blob metadata
+
+                Module: {blob['module']}
+                Blob:   {blob['path']}
+                URL:    {blob['url']}
+                Info:   {blob['description']}'''))
+
     def fetch(self, args):
         blobs = self.get_blobs(args)
         for blob in blobs:
-            if blob['status'] == 'A':
+            if blob['status'] == zephyr_module.BLOB_PRESENT:
                 log.dbg('Blob {module}: {abspath} is up to date'.format(**blob))
                 continue
             log.inf('Fetching blob {module}: {abspath}'.format(**blob))
             self.fetch_blob(blob['url'], blob['abspath'])
+            self.verify_blob(blob)
 
     def clean(self, args):
         blobs = self.get_blobs(args)
         for blob in blobs:
-            if blob['status'] == 'D':
+            if blob['status'] == zephyr_module.BLOB_NOT_PRESENT:
                 log.dbg('Blob {module}: {abspath} not in filesystem'.format(**blob))
                 continue
             log.inf('Deleting blob {module}: {status} {abspath}'.format(**blob))

scripts/zephyr_module.py

@@ -345,7 +345,7 @@ def kconfig_snippet(meta, path, kconfig_file=None, blobs=False, sysbuild=False):
 
 def process_kconfig(module, meta):
     blobs = process_blobs(module, meta)
-    taint_blobs = len(tuple(filter(lambda b: b['status'] != 'D', blobs))) != 0
+    taint_blobs = any(b['status'] != BLOB_NOT_PRESENT for b in blobs)
     section = meta.get('build', dict())
     module_path = PurePath(module)
     module_yml = module_path.joinpath('zephyr/module.yml')