handle SSL errors for self-signed certificates in SSEClient and DAQWorker

src/aare/gui/threads/daq_worker.py

@@ -10,7 +10,7 @@ from collections import deque
 from typing import cast, Literal
 
 from PySide6.QtCore import Signal, QUrl, Slot, QTimer, QObject, QByteArray
-from PySide6.QtNetwork import QNetworkAccessManager, QNetworkRequest, QNetworkReply
+from PySide6.QtNetwork import QNetworkAccessManager, QNetworkRequest, QNetworkReply, QSslError
 from jfjoch_client import ScanResult, ScanResultImagesInner
 
 from aare.common.auth_models import BatonStatus
@@ -120,6 +120,7 @@ class DAQWorker(QObject):
         self.__token = token
         self.__base_url = base_url
         self.__net_manager = QNetworkAccessManager()
+        self.__net_manager.sslErrors.connect(self._handle_ssl_errors)
         self.__timer = QTimer()
         self.__timer.setInterval(500)
         self.__timer.timeout.connect(self.regular_update)
@@ -326,6 +327,18 @@ class DAQWorker(QObject):
         reply = self.__net_manager.get(request)
         reply.finished.connect(lambda: self.handle_status_response(reply))
 
+    _SELF_SIGNED_ERRORS = {
+        QSslError.SslError.SelfSignedCertificate,
+        QSslError.SslError.SelfSignedCertificateInChain,
+    }
+
+    @staticmethod
+    def _handle_ssl_errors(reply: QNetworkReply, errors: list):
+        self_signed = [e for e in errors if e.error() in DAQWorker._SELF_SIGNED_ERRORS]
+        if self_signed:
+            reply.ignoreSslErrors(self_signed)
+
+
     @staticmethod
     def handle_response(reply: QNetworkReply):
         """

src/aare/gui/threads/sse_client.py

@@ -1,6 +1,6 @@
 from typing import Optional, Dict
 from PySide6.QtCore import QObject, Signal, Slot, QUrl, QTimer, QByteArray
-from PySide6.QtNetwork import QNetworkAccessManager, QNetworkRequest, QNetworkReply
+from PySide6.QtNetwork import QNetworkAccessManager, QNetworkRequest, QNetworkReply, QSslError
 
 
 class SSEClient(QObject):
@@ -14,6 +14,7 @@ class SSEClient(QObject):
     def __init__(self, parent=None):
         super().__init__(parent)
         self._network_manager = QNetworkAccessManager(self)
+        self._network_manager.sslErrors.connect(self._handle_ssl_errors)
         self._reply: Optional[QNetworkReply] = None
         self._reconnect_timer = QTimer(self)
         self._reconnect_timer.setSingleShot(True)
@@ -60,6 +61,17 @@ class SSEClient(QObject):
         """Check if connected to SSE"""
         return self._connected and self._reply and self._reply.isOpen()
 
+    _SELF_SIGNED_ERRORS = {
+        QSslError.SslError.SelfSignedCertificate,
+        QSslError.SslError.SelfSignedCertificateInChain,
+    }
+
+    @staticmethod
+    def _handle_ssl_errors(reply: QNetworkReply, errors: list):
+        self_signed = [e for e in errors if e.error() in SSEClient._SELF_SIGNED_ERRORS]
+        if self_signed:
+            reply.ignoreSslErrors(self_signed)
+
     def _start_connection(self):
         """Start SSE connection"""
         request = QNetworkRequest(self._url)