Add runtime warning if the MCU firmware is out of date, based on a hash of the firmware sources

.gitignore

@@ -5,6 +5,7 @@ out
 __pycache__/
 .config
 .config.old
+.sources
 klippy/.version
 .history/
 .DS_Store

docs/Config_Reference.md

@@ -123,7 +123,9 @@ A collection of Kalico-specific system options
 #endstop_sample_count: 4
 #   How many times we should check the endstop state when homing
 #   Unless your endstop is noisy and unreliable, you should be able to lower this to 1
-
+#warn_on_mismatched_firmware_sources: True
+#   Warn when the mcu firmware does not match the firmware sources for the current
+#   Kalico version
 
 # Logging options:
 

klippy/extras/danger_options.py

@@ -37,6 +37,9 @@ def __init__(self, config):
         self.homing_elapsed_distance_tolerance = config.getfloat(
             "homing_elapsed_distance_tolerance", 0.5, minval=0.0
         )
+        self.warn_on_mismatched_firmware_sources = config.getboolean(
+            "warn_on_mismatched_firmware_sources", True
+        )
 
         temp_ignore_limits = False
         if config.getboolean("temp_ignore_limits", None) is None:

klippy/mcu.py

@@ -1191,6 +1191,7 @@ def _mcu_identify(self):
                 raise error(str(e))
         if get_danger_options().log_startup_info:
             logging.info(self._log_info())
+        pconfig = self._printer.lookup_object("configfile")
         ppins = self._printer.lookup_object("pins")
         pin_resolver = ppins.get_pin_resolver(self._name)
         for cname, value in self.get_constants().items():
@@ -1214,16 +1215,26 @@ def _mcu_identify(self):
             )
         app = msgparser.get_app_info()
         version, build_versions = msgparser.get_version_info()
+        sources_hash = msgparser.get_sources_hash()
         self._get_status_info["app"] = app
         self._get_status_info["mcu_version"] = version
         self._get_status_info["mcu_build_versions"] = build_versions
+        self._get_status_info["mcu_sources_hash"] = sources_hash
         self._get_status_info["mcu_constants"] = msgparser.get_constants()
         if app in ("Klipper", "Danger-Klipper"):
-            pconfig = self._printer.lookup_object("configfile")
             pconfig.runtime_warning(
                 f"MCU {self._name!r} currently has firmware compiled for {app} (version {version})."
                 f" It is recommended to re-flash for best compatiblity with Kalico"
             )
+        elif (
+            get_danger_options().warn_on_mismatched_firmware_sources
+            and sources_hash
+            != self._printer.get_start_args().get("sources_hash")
+        ):
+            pconfig.runtime_warning(
+                f"MCU {self._name!r} firmware is out of date and may not function as intended."
+                f" Please re-flash as soon as possible"
+            )
         self.register_response(self._handle_shutdown, "shutdown")
         self.register_response(self._handle_shutdown, "is_shutdown")
         self.register_response(self._handle_mcu_stats, "stats")

klippy/msgproto.py

@@ -488,6 +488,7 @@ def process_identify(self, data, decompress=True):
             self.app = data.get("app", "")
             self.version = data.get("version", "")
             self.build_versions = data.get("build_versions", "")
+            self.sources_hash = data.get("sources_hash", "unknown")
         except error as e:
             raise
         except Exception as e:
@@ -503,6 +504,9 @@ def get_app_info(self):
     def get_version_info(self):
         return self.version, self.build_versions
 
+    def get_sources_hash(self):
+        return self.sources_hash
+
     def get_messages(self):
         return list(self.messages)
 

klippy/printer.py

@@ -787,15 +787,18 @@ def main():
     logging.info("=======================")
     logging.info("Starting Klippy...")
     git_info = util.get_git_version()
+    sources_hash = util.get_firmware_hash()
     git_vers = git_info["version"]
 
     extra_git_desc = ""
     extra_git_desc += "\nBranch: %s" % (git_info["branch"])
     extra_git_desc += "\nRemote: %s" % (git_info["remote"])
     extra_git_desc += "\nTracked URL: %s" % (git_info["url"])
+    extra_git_desc += "\nSources Hash: %s" % (sources_hash)
     start_args["software_version"] = git_vers
     start_args["git_branch"] = git_info["branch"]
     start_args["git_remote"] = git_info["remote"]
+    start_args["sources_hash"] = sources_hash
     start_args["cpu_info"] = util.get_cpu_info()
     if bglogger is not None:
         versions = "\n".join(

klippy/util.py

@@ -4,9 +4,11 @@
 #
 # This file may be distributed under the terms of the GNU GPLv3 license.
 import fcntl
+import hashlib
 import json
 import logging
 import os
+import pathlib
 import pty
 import signal
 import subprocess
@@ -252,3 +254,24 @@ def get_git_version(from_file=True):
     if from_file:
         git_info["version"] = get_version_from_file(klippy_src)
     return git_info
+
+
+def get_firmware_hash():
+    root_dir = pathlib.Path(__file__).parent.parent
+    hash_cache = root_dir / ".sources"
+    source_files = sorted(
+        file
+        for path in (root_dir / "src", root_dir / "lib")
+        for file in path.glob("**/*")
+        if file.is_file()
+    )
+    last_modified = max(file.stat().st_mtime for file in source_files)
+    if hash_cache.is_file() and hash_cache.stat().st_mtime >= last_modified:
+        return hash_cache.read_text()
+    hash = hashlib.blake2b(digest_size=16, usedforsecurity=False)
+    for file in source_files:
+        hash.update(b"\x00" + bytes(file.relative_to(root_dir)) + b"\x00")
+        hash.update(file.read_bytes())
+    digest = hash.hexdigest()
+    hash_cache.write_text(digest)
+    return digest

scripts/buildcommands.py

@@ -21,6 +21,7 @@
 sys.path.insert(0, str(pathlib.Path(__file__).parent.parent / "klippy"))
 
 import msgproto
+from util import get_firmware_hash
 
 FILEHEADER = """
 /* DO NOT EDIT! This is an autogenerated file. See scripts/buildcommands.py. */
@@ -644,21 +645,25 @@ def tool_versions(tools):
 class HandleVersions:
     def __init__(self):
         self.ctr_dispatch = {}
-        self.toolstr = self.version = ""
+        self.toolstr = self.version = self.sources = ""
 
     def update_data_dictionary(self, data):
         data["version"] = self.version
         data["build_versions"] = self.toolstr
         data["app"] = "Kalico"
         data["license"] = "GNU GPLv3"
+        data["sources_hash"] = self.sources
 
     def generate_code(self, options):
         cleanbuild, self.toolstr = tool_versions(options.tools)
         self.version = build_version(options.extra, cleanbuild)
+        self.sources = get_firmware_hash()
         sys.stdout.write("Version: %s\n" % (self.version,))
-        return "\n// version: %s\n// build_versions: %s\n" % (
+        sys.stdout.write("Sources Hash: %s\n" % (self.sources,))
+        return "\n// version: %s\n// build_versions: %s\n// sources: %s\n" % (
             self.version,
             self.toolstr,
+            self.sources,
         )