Show the user a warning when the printer requires authentication.

Also hide it when appropriate.

part of CURA-12717
This commit is contained in:
Remco Burema 2025-10-22 12:19:26 +02:00
parent 6f41127628
commit cd3aa02440
4 changed files with 65 additions and 4 deletions

View file

@ -0,0 +1,34 @@
# Copyright (c) 2025 UltiMaker
# Cura is released under the terms of the LGPLv3 or higher.
from UM.Message import Message
class AuthorizationRequiredMessage:
_inner_message_instance = None
class InnerMessage(Message):
def __init__(self, printer_name: str, err_message: str) -> None:
super().__init__(
text = printer_name,
title = err_message,
message_type = Message.MessageType.WARNING,
lifetime = 0
)
@classmethod
def _getInstance(cls) -> Message:
if cls._inner_message_instance is None:
cls._inner_message_instance = cls.InnerMessage("", "")
return cls._inner_message_instance
@classmethod
def show(cls, printer_name: str, err_message: str) -> None:
msg = cls._getInstance()
msg.setText(printer_name)
msg.setTitle(err_message)
msg.show()
@classmethod
def hide(cls) -> None:
cls._getInstance().hide()

View file

@ -16,6 +16,7 @@ from UM.Logger import Logger
from cura.CuraApplication import CuraApplication
from ..Messages.AuthorizationRequiredMessage import AuthorizationRequiredMessage
from ..Models.BaseModel import BaseModel
from ..Models.Http.ClusterPrintJobStatus import ClusterPrintJobStatus
from ..Models.Http.ClusterPrinterStatus import ClusterPrinterStatus
@ -56,7 +57,7 @@ class ClusterApiClient:
# In order to avoid garbage collection we keep the callbacks in this list.
_anti_gc_callbacks = [] # type: List[Callable[[], None]]
def __init__(self, address: str, on_error: Callable) -> None:
def __init__(self, address: str, on_error: Callable, on_auth_required: Callable) -> None:
"""Initializes a new cluster API client.
:param address: The network address of the cluster to call.
@ -68,6 +69,7 @@ class ClusterApiClient:
self._on_error = on_error
self._auth_tries = 0
self._on_auth_required = on_auth_required
prefs = CuraApplication.getInstance().getPreferences()
prefs.addPreference("cluster_api/auth_ids", "{}")
@ -305,6 +307,7 @@ class ClusterApiClient:
self._auth_id = None
self._auth_key = None
self._on_auth_required(reply.errorString())
nonce_match = re.search(r'nonce="([^"]+)', str(reply.rawHeader(b"WWW-Authenticate")))
if nonce_match:
self._nonce = nonce_match.group(1)
@ -312,9 +315,13 @@ class ClusterApiClient:
self._setLocalValueToPrefDict("cluster_api/nonce_counts", self._nonce_count)
self._setLocalValueToPrefDict("cluster_api/nonces", self._nonce)
CuraApplication.getInstance().savePreferences()
self._on_error(reply.errorString())
else:
self._on_error(reply.errorString())
return
if self._auth_id and self._auth_key and self._nonce_count > 1:
AuthorizationRequiredMessage.hide()
# If no parse model is given, simply return the raw data in the callback.
if not model:
on_finished(reply.readAll())

View file

@ -21,6 +21,7 @@ from cura.PrinterOutput.PrinterOutputDevice import ConnectionType
from .ClusterApiClient import ClusterApiClient
from .SendMaterialJob import SendMaterialJob
from ..ExportFileJob import ExportFileJob
from ..Messages.AuthorizationRequiredMessage import AuthorizationRequiredMessage
from ..UltimakerNetworkedPrinterOutputDevice import UltimakerNetworkedPrinterOutputDevice
from ..Messages.PrintJobUploadBlockedMessage import PrintJobUploadBlockedMessage
from ..Messages.PrintJobUploadErrorMessage import PrintJobUploadErrorMessage
@ -239,9 +240,19 @@ class LocalClusterOutputDevice(UltimakerNetworkedPrinterOutputDevice):
if print_job.getPreviewImage() is None:
self.getApiClient().getPrintJobPreviewImage(print_job.key, print_job.updatePreviewImageData)
def _onAuthRequired(self, error_msg: str) -> None:
active_name = CuraApplication.getInstance().getOutputDeviceManager().getActiveDevice().getName()
if self._name == active_name:
Logger.info(f"Authorization required for {self._name}: {error_msg}")
AuthorizationRequiredMessage.show(self._name, error_msg)
def getApiClient(self) -> ClusterApiClient:
"""Get the API client instance."""
if not self._cluster_api:
self._cluster_api = ClusterApiClient(self.address, on_error = lambda error: Logger.log("e", str(error)))
self._cluster_api = ClusterApiClient(
self._address,
on_error = lambda error: Logger.log("e", str(error)),
on_auth_required = self._onAuthRequired
)
return self._cluster_api

View file

@ -15,6 +15,7 @@ from cura.Settings.GlobalStack import GlobalStack
from .ZeroConfClient import ZeroConfClient
from .ClusterApiClient import ClusterApiClient
from .LocalClusterOutputDevice import LocalClusterOutputDevice
from ..Messages.AuthorizationRequiredMessage import AuthorizationRequiredMessage
from ..UltimakerNetworkedPrinterOutputDevice import UltimakerNetworkedPrinterOutputDevice
from ..Messages.CloudFlowMessage import CloudFlowMessage
from ..Messages.LegacyDeviceNoLongerSupportedMessage import LegacyDeviceNoLongerSupportedMessage
@ -44,6 +45,7 @@ class LocalClusterOutputDeviceManager:
# Persistent dict containing the networked clusters.
self._discovered_devices = {} # type: Dict[str, LocalClusterOutputDevice]
self._output_device_manager = CuraApplication.getInstance().getOutputDeviceManager()
self._output_device_manager.activeDeviceChanged.connect(self._onActiveDeviceChanged)
# Hook up ZeroConf client.
self._zero_conf_client = ZeroConfClient()
@ -69,10 +71,17 @@ class LocalClusterOutputDeviceManager:
self.stop()
self.start()
def _onActiveDeviceChanged(self):
AuthorizationRequiredMessage.hide()
def _onAuthRequired(self, error_msg: str) -> None:
Logger.info(f"Authorization required: {error_msg}")
AuthorizationRequiredMessage.show("", error_msg)
def addManualDevice(self, address: str, callback: Optional[Callable[[bool, str], None]] = None) -> None:
"""Add a networked printer manually by address."""
api_client = ClusterApiClient(address, lambda error: Logger.log("e", str(error)))
api_client = ClusterApiClient(address, lambda error: Logger.log("e", str(error)), self._onAuthRequired)
api_client.getSystem(lambda status: self._onCheckManualDeviceResponse(address, status, callback))
def removeManualDevice(self, device_id: str, address: Optional[str] = None) -> None: