Switch to 256-bit ECDSA algorithm (#123)

* label worker threads

* remove image transform (included in input op)

* fix surprising logging behavior

* fixup mqtt client init flow

* finish upgrading to es256 encryption

octoprint_nanny/clients/mqtt.py

@@ -9,17 +9,18 @@
 import io
 import random
 import paho.mqtt.client as mqtt
+from typing import List
 
 import beeline
 
 from octoprint_nanny.utils.encoder import NumpyEncoder
 from octoprint_nanny.clients.honeycomb import HoneycombTracer
 
 
-JWT_EXPIRES_MINUTES = os.environ.get("OCTOPRINT_NANNY_MQTT_JWT_EXPIRES_MINUTES", 600)
+JWT_EXPIRES_MINUTES = os.environ.get("OCTOPRINT_NANNY_MQTT_JWT_EXPIRES_MINUTES", 1380)
 GCP_PROJECT_ID = os.environ.get("OCTOPRINT_NANNY_GCP_PROJECT_ID", "print-nanny")
 MQTT_BRIDGE_HOSTNAME = os.environ.get(
-    "OCTOPRINT_NANNY_MQTT_BRIDGE_HOSTNAME", "mqtt.googleapis.com"
+    "OCTOPRINT_NANNY_MQTT_BRIDGE_HOSTNAME", "mqtt.2030.ltsapis.goog"
 )
 
 MQTT_BRIDGE_PORT = os.environ.get("OCTOPRINT_NANNY_MQTT_BRIDGE_PORT", 443)
@@ -54,8 +55,8 @@ def __init__(
         device_id: str,
         device_cloudiot_id: str,
         private_key_file: str,
-        ca_certs,
-        algorithm="RS256",
+        ca_cert: str,
+        algorithm="ES256",
         mqtt_receive_queue=None,
         mqtt_bridge_hostname=MQTT_BRIDGE_HOSTNAME,
         mqtt_bridge_port=MQTT_BRIDGE_PORT,
@@ -69,8 +70,7 @@ def __init__(
         project_id=GCP_PROJECT_ID,
         region=IOT_DEVICE_REGISTRY_REGION,
         registry_id=IOT_DEVICE_REGISTRY,
-        tls_version=ssl.PROTOCOL_TLS,
-        keepalive=900,  # 15 minutes
+        keepalive=900,
         trace_context={},
         message_callbacks=[],  # see message_callback_add() https://www.eclipse.org/paho/index.php?page=clients/python/docs/index.php#subscribe-unsubscribe
     ):
@@ -84,19 +84,19 @@ def __init__(
         self.project_id = project_id
         self.mqtt_bridge_hostname = mqtt_bridge_hostname
         self.mqtt_bridge_port = mqtt_bridge_port
-        self.ca_certs = ca_certs
         self.region = region
         self.registry_id = registry_id
         self.keepalive = keepalive
 
-        self.tls_version = tls_version
+        self.ca_cert = ca_cert
+
         self.region = region
         self.algorithm = algorithm
 
         self.mqtt_receive_queue = mqtt_receive_queue
         self._honeycomb_tracer = HoneycombTracer(service_name="octoprint_plugin")
 
-        self.client = mqtt.Client(client_id=client_id, protocol=mqtt.MQTTv311)
+        self.client = mqtt.Client(client_id=client_id)
         logger.info(f"Initializing MQTTClient from {locals()}")
 
         # register callback functions
@@ -131,9 +131,6 @@ def __init__(
             self.mqtt_default_telemetry_topic, BOUNDING_BOX_EVENT_FOLDER
         )
 
-        # configure tls
-        self.client.tls_set(ca_certs=ca_certs, tls_version=tls_version)
-
     ###
     #   callbacks
     ##
@@ -241,10 +238,16 @@ def _on_disconnect(self, client, userdata, rc):
                 time.sleep(delay)
                 if minimum_backoff_time <= MAXIMUM_BACKOFF_TIME:
                     minimum_backoff_time *= 2
-            self.connect()
+            self.client.reconnect()
 
     @beeline.traced("MQTTClient.connect")
     def connect(self):
+        # configure tls
+        self.client.tls_set(
+            ca_certs=self.ca_cert,
+            ciphers="ECDHE-RSA-AES128-GCM-SHA256",
+            tls_version=ssl.PROTOCOL_TLS,
+        )
         self.client.username_pw_set(
             username="unused",
             password=create_jwt(self.project_id, self.private_key_file, self.algorithm),
@@ -307,17 +310,15 @@ def create_jwt(
      algorithm: The encryption algorithm to use. Either 'RS256' or 'ES256'
     Returns:
         A JWT generated from the given project_id and private key, which
-        expires in 20 minutes. After 20 minutes, your client will be
+        expires in JWT_EXPIRES_MINUTES minutes. After JWT_EXPIRES_MINUTES minutes, your client will be
         disconnected, and a new JWT will have to be generated.
     Raises:
         ValueError: If the private_key_file does not contain a known key.
     """
-    _jwt = jwt.JWT()
 
-    exp = jwt.utils.get_int_from_datetime(
-        datetime.utcnow() + timedelta(minutes=jwt_expires_minutes)
-    )
-    iat = jwt.utils.get_int_from_datetime(datetime.utcnow())
+    exp = datetime.utcnow() + timedelta(minutes=jwt_expires_minutes)
+
+    iat = datetime.utcnow()
     token = {
         # The time that the token was issued at
         "iat": iat,
@@ -329,12 +330,12 @@ def create_jwt(
 
     # Read the private key file.
     with open(private_key_file, "rb") as f:
-        signing_key = jwt.jwk_from_pem(f.read())
+        signing_key = f.read()
 
     logger.info(
         "Creating JWT using {} from private key file {}".format(
             algorithm, private_key_file
         )
     )
 
-    return _jwt.encode(token, signing_key, alg=algorithm)
+    return jwt.encode(token, signing_key, algorithm=algorithm)

octoprint_nanny/manager.py

@@ -109,19 +109,16 @@ def _register_plugin_event_handlers(self):
         """
         Events.PLUGIN_OCTOPRINT_NANNY* events are not available on Events until plugin is fully initialized
         """
-        self.mqtt_manager.publisher_worker.register_callbacks(
-            {
-                Events.PLUGIN_OCTOPRINT_NANNY_MONITORING_START: self.monitoring_manager.start,
-                Events.PLUGIN_OCTOPRINT_NANNY_MONITORING_STOP: self.monitoring_manager.stop,
-            }
-        )
 
-        self.mqtt_manager.subscriber_worker.register_callbacks(
-            {
-                Events.PLUGIN_OCTOPRINT_NANNY_MONITORING_START: self.monitoring_manager.start,
-                Events.PLUGIN_OCTOPRINT_NANNY_MONITORING_STOP: self.monitoring_manager.stop,
-            }
-        )
+        callbacks = {
+            Events.PLUGIN_OCTOPRINT_NANNY_MONITORING_START: self.monitoring_manager.start,
+            Events.PLUGIN_OCTOPRINT_NANNY_MONITORING_STOP: self.monitoring_manager.stop,
+        }
+        self.mqtt_manager.publisher_worker.register_callbacks(callbacks)
+        logger.info(f"Registered callbacks {callbacks} on publisher worker")
+
+        self.mqtt_manager.subscriber_worker.register_callbacks(callbacks)
+        logger.info(f"Registered callbacks {callbacks} on subscriber worker")
 
     @beeline.traced("WorkerManager.on_settings_initialized")
     def on_settings_initialized(self):
@@ -146,31 +143,28 @@ def apply_auth(self):
         self.mqtt_manager.start()
 
     @beeline.traced("WorkerManager.apply_monitoring_settings")
-    def apply_monitoring_settings(self):
+    async def apply_monitoring_settings(self):
 
         self.plugin.settings.reset_monitoring_settings()
         logger.info(
             "Stopping any existing monitoring processes to apply new calibration"
         )
-        self.monitoring_manager.stop()
+        await self.monitoring_manager.stop()
         if self.monitoring_active:
             logger.info(
                 "Monitoring was active when new calibration was applied. Re-initializing monitoring processes"
             )
-            self.monitoring_manager.start()
+            await self.monitoring_manager.start()
 
     @beeline.traced("WorkerManager.shutdown")
-    def shutdown(self):
-        self.monitoring_manager.stop()
+    async def shutdown(self):
+        await self.monitoring_manager.stop()
 
-        asyncio.run_coroutine_threadsafe(
-            self.plugin.settings.rest_client.update_octoprint_device(
-                self.plugin.settings.device_id, monitoring_active=False
-            ),
-            self.loop,
-        ).result()
+        await self.plugin.settings.rest_client.update_octoprint_device(
+            self.plugin.settings.device_id, monitoring_active=False
+        )
 
-        self.mqtt_manager.stop()
+        await self.mqtt_manager.stop()
         self._honeycomb_tracer.on_shutdown()
 
     @beeline.traced("WorkerManager.on_print_start")