Initial commit

Author: foosel

.editorconfig

@@ -0,0 +1,17 @@
+# This file is for unifying the coding style for different editors and IDEs
+# editorconfig.org
+
+root = true
+
+[*]
+end_of_line = lf
+charset = utf-8
+insert_final_newline = true
+trim_trailing_whitespace = true
+
+[**.py]
+indent_style = tab
+
+[**.js]
+indent_style = space
+indent_size = 4

.gitignore

@@ -0,0 +1,9 @@
+*.pyc
+*.swp
+.idea
+*.iml
+build
+dist
+*.egg*
+.DS_Store
+*.zip

MANIFEST.in

@@ -0,0 +1,4 @@
+include README.md
+recursive-include octoprint_firmware_check/templates *
+recursive-include octoprint_firmware_check/translations *
+recursive-include octoprint_firmware_check/static *

README.md

@@ -0,0 +1,51 @@
+# Firmware Check
+
+The Firmware Check plugin tries to identify printers or rather printer firmware with known safety issues, such as
+disabled thermal runaway protection, or other kinds of severe issues, like known communication crippling bugs, and
+displays a warning box to logged in users on identification of such a firmware.
+
+![Screenshot](extras/screenshot.png)
+
+It was formerly called "Printer Safety Check" and used to be bundled with OctoPrint since version
+1.3.7. It was unbundled in 1.4.1 and turned into an install dependency to allow for a separate release cycle. It is
+still considered a core plugin of OctoPrint, treated as if bundled and thus also active in [safe mode](https://docs.octoprint.org/en/master/features/safemode.html).
+
+## Setup
+
+The plugin is part of the core dependencies of OctoPrint 1.4.1+ and will be installed automatically alongside it.
+
+In case you want to manually install it into an older version for whatever reason, install via the bundled
+[Plugin Manager](https://docs.octoprint.org/en/master/bundledplugins/pluginmanager.html)
+or manually using this URL:
+
+    https://github.com/OctoPrint/OctoPrint-FirmwareCheck/archive/master.zip
+
+## Events
+
+### plugin_firmware_check_warning
+
+*(as `Events.PLUGIN_FIRMWARE_CHECK_WARNING`)*
+
+A firmware check warning was triggered.
+
+Payload:
+  * `warning_type`: type of warning that was triggered (currently `firmware-unsafe` or `firmware-broken`)
+  * `check_name`: name of check that was triggered (e.g. `aneta8`, `cbd`)
+  * `check_type`: type of check that was triggered (e.g. `m115`, `received` or `cap`)
+
+## Detected issues
+
+The plugin currently issues two types of warnings: `firmware-unsafe` for firmware known to have severe safety issues
+and `firmware-broken` for firmware known to have a broken implementation of the communication protocol.
+
+### Unsafe firmware
+
+Please refer to the [entry on the "unsafe firmware" warning in OctoPrint's FAQ](https://faq.octoprint.org/warning-firmware-unsafe)
+for a list of currently identified printers.
+
+### Broken firmware
+
+#### "CBD" firmware
+
+Please refer to the [entry on the "broken CBD firmware" warning in OctoPrint's FAQ](https://faq.octoprint.org/warning-firmware-broken-cbd)
+for a list of currently identified printers.

babel.cfg

@@ -0,0 +1,6 @@
+[python: */**.py]
+[jinja2: */**.jinja2]
+extensions=jinja2.ext.autoescape, jinja2.ext.with_
+
+[javascript: */**.js]
+extract_messages = gettext, ngettext

extras/screenshot.png

@@ -0,0 +1,213 @@
+# -*- coding: utf-8 -*-
+from __future__ import absolute_import, division, print_function, unicode_literals
+
+__license__ = 'GNU Affero General Public License http://www.gnu.org/licenses/agpl.html'
+__copyright__ = "Copyright (C) 2018 The OctoPrint Project - Released under terms of the AGPLv3 License"
+
+import octoprint.plugin
+
+from octoprint.events import Events
+from octoprint.access import USER_GROUP, ADMIN_GROUP
+from octoprint.access.permissions import Permissions
+from octoprint.util import to_unicode
+
+from .checks import Severity
+from .checks.firmware_unsafe import FirmwareUnsafeChecks
+from .checks.firmware_broken import FirmwareBrokenChecks
+
+import flask
+from flask_babel import gettext
+
+import textwrap
+
+TERMINAL_WARNING = """
+!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
+{message}
+
+Learn more at {url}
+!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
+
+"""
+
+FIRMWARE_CHECKS = {
+	"firmware-unsafe": FirmwareUnsafeChecks.as_dict(),
+	"firmware-broken": FirmwareBrokenChecks.as_dict()
+}
+
+class FirmwareCheckPlugin(octoprint.plugin.AssetPlugin,
+                          octoprint.plugin.EventHandlerPlugin,
+                          octoprint.plugin.SimpleApiPlugin,
+                          octoprint.plugin.TemplatePlugin):
+
+	# noinspection PyMissingConstructor
+	def __init__(self):
+		self._warnings = dict()
+		self._scan_received = True
+
+	##~~ TemplatePlugin API
+
+	def get_template_configs(self):
+		return [
+			dict(type="sidebar",
+			     name=gettext("Attention!"),
+			     data_bind="visible: printerState.isOperational() && loginState.isAdmin() && warnings().length > 0",
+			     icon="exclamation-triangle",
+			     styles_wrapper=["display: none"])
+		]
+
+	##~~ AssetPlugin API
+
+	def get_assets(self):
+		return dict(js=("js/firmware_check.js",),
+		            clientjs=("clientjs/firmware_check.js",),
+		            css=("css/firmware_check.css",),
+		            less=("less/firmware_check.less",))
+
+	##~~ EventHandlerPlugin API
+
+	def on_event(self, event, payload):
+		if event == Events.DISCONNECTED:
+			self._reset_warnings()
+			self._reset_state()
+			self._reset_checks()
+
+	##~~ SimpleApiPlugin API
+
+	def on_api_get(self, request):
+		if not Permissions.PLUGIN_FIRMWARE_CHECK_DISPLAY.can():
+			return flask.make_response("Insufficient rights", 403)
+		return flask.jsonify(self._warnings)
+
+	##~~ GCODE received hook handler
+
+	def on_gcode_received(self, comm_instance, line, *args, **kwargs):
+		if self._scan_received:
+			self._run_checks("received", to_unicode(line, errors="replace"))
+		return line
+
+	##~~ Firmware info hook handler
+
+	def on_firmware_info_received(self, comm_instance, firmware_name, firmware_data):
+		self._run_checks("m115",
+		                 to_unicode(firmware_name, errors="replace"),
+		                 dict((to_unicode(key, errors="replace"), to_unicode(value, errors="replace"))
+		                      for key, value in firmware_data.items()))
+		self._scan_received = False
+
+	##~~ Firmware capability hook handler
+
+	def on_firmware_cap_received(self, comm_instance, cap, enabled, all_caps):
+		self._run_checks("cap",
+		                 to_unicode(cap, errors="replace"),
+		                 enabled)
+
+	##~~ Additional permissions hook handler
+
+	def get_additional_permissions(self):
+		return [
+			dict(key="DISPLAY",
+			     name="Display firmware check warnings",
+			     description=gettext("Allows to see firmware check warnings"),
+			     roles=["display"],
+			     default_groups=[USER_GROUP])
+		]
+
+	##~~ Helpers
+
+	def _run_checks(self, check_type, *args, **kwargs):
+		changes = False
+
+		for warning_type, check_data in FIRMWARE_CHECKS.items():
+			checks = check_data.get("checks")
+			message = check_data.get("message")
+			severity = check_data.get("severity", Severity.CRITICAL)
+			url = "https://faq.octoprint.org/warning-{warning_type}".format(warning_type=warning_type)
+			if not checks or not message:
+				continue
+
+			for check in checks:
+				if not check.active:
+					# skip non active checks
+					continue
+
+				method = getattr(check, check_type, None)
+				if not callable(method):
+					# skip uncallable checks
+					continue
+
+				# execute method
+				try:
+					method(*args, **kwargs)
+				except Exception:
+					self._logger.exception("There was an error running method {} on check {!r}".format(check_type, check))
+					continue
+
+				# check if now triggered
+				if check.triggered:
+					if check.url is not None:
+						url = check.url
+
+					self._register_warning(warning_type, message, severity, url)
+
+					# noinspection PyUnresolvedReferences
+					self._event_bus.fire(Events.PLUGIN_FIRMWARE_CHECK_WARNING, dict(check_name=check.name,
+					                                                                warning_type=warning_type,
+					                                                                severity=severity,
+					                                                                url=url))
+					changes = True
+					break
+
+		if changes:
+			self._ping_clients()
+
+	def _register_warning(self, warning_type, message, severity, url):
+		self._log_to_terminal(TERMINAL_WARNING.format(message="\n".join(textwrap.wrap(message, 75)),
+		                                              warning_type=warning_type,
+		                                              url=url))
+		self._warnings[warning_type] = dict(message=message,
+		                                    severity=severity,
+		                                    url=url)
+
+	def _reset_warnings(self):
+		self._warnings.clear()
+		self._ping_clients()
+
+	def _reset_state(self):
+		self._scan_received = True
+
+	def _reset_checks(self):
+		for warning_type, check_data in FIRMWARE_CHECKS.items():
+			checks = check_data.get("checks")
+			if not checks:
+				continue
+
+			for check in checks:
+				check.reset()
+
+	def _log_to_terminal(self, message):
+		if self._printer:
+			lines = message.split("\n")
+			self._printer.log_lines(*lines)
+
+	def _ping_clients(self):
+		self._plugin_manager.send_plugin_message(self._identifier, dict(type="update"))
+
+
+def register_custom_events(*args, **kwargs):
+	return ["warning",]
+
+
+__plugin_name__ = "Firmware Check"
+__plugin_pythoncompat__ = ">=2.7,<4" # python 2 and 3
+__plugin_disabling_discouraged__ = gettext("Without this plugin OctoPrint will no longer be able to "
+                                           "check if the printer it is connected to has a known safety "
+                                           "issue or otherwise broken firmware and inform you about that fact.")
+__plugin_implementation__ = FirmwareCheckPlugin()
+__plugin_hooks__ = {
+	"octoprint.comm.protocol.gcode.received": (__plugin_implementation__.on_gcode_received, 100),
+	"octoprint.comm.protocol.firmware.info": (__plugin_implementation__.on_firmware_info_received, 100),
+	"octoprint.comm.protocol.firmware.capabilities": (__plugin_implementation__.on_firmware_cap_received, 100),
+	"octoprint.events.register_custom_events": register_custom_events,
+	"octoprint.access.permissions": __plugin_implementation__.get_additional_permissions
+}
+

octoprint_firmware_check/__init__.py

@@ -0,0 +1,60 @@
+# -*- coding: utf-8 -*-
+from __future__ import absolute_import, division, print_function, unicode_literals
+
+__license__ = 'GNU Affero General Public License http://www.gnu.org/licenses/agpl.html'
+__copyright__ = "Copyright (C) 2019 The OctoPrint Project - Released under terms of the AGPLv3 License"
+
+
+class Check(object):
+	name = None
+	url = None
+
+	def __init__(self):
+		self._active = True
+		self._triggered = False
+
+	def received(self, line):
+		"""Called when receiving a new line from the printer"""
+		pass
+
+	def m115(self, name, data):
+		"""Called when receiving the response to an M115 from the printer"""
+		pass
+
+	def cap(self, cap, enabled):
+		"""Called when receiving a capability report line"""
+		pass
+
+	@property
+	def active(self):
+		"""Whether this check is still active"""
+		return self._active
+
+	@property
+	def triggered(self):
+		"""Whether the check has been triggered"""
+		return self._triggered
+
+	def reset(self):
+		self._active = True
+		self._triggered = False
+
+
+class AuthorCheck(Check):
+	authors = ()
+
+	AUTHOR = "| Author: ".lower()
+
+	def received(self, line):
+		if not line:
+			return
+
+		lower_line = line.lower()
+		if self.AUTHOR in lower_line:
+			self._triggered = any(map(lambda x: x in lower_line, self.authors))
+			self._active = False
+
+
+class Severity(object):
+	INFO = "info"
+	CRITICAL = "critical"