refactor: ZabbixAPI exception handling

CHANGELOG

@@ -13,6 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - `create_notification_user`: Now adds users to the default user group in addition to the notification user group to match behavior in V2.
 - `show_media_types`: Now shows the formatted string representation of the media type `type` field instead of an integer.
+- Auth tokens and passwords from API request errors are now masked by default in output.
 
 ### Deprecated
 

zabbix_cli/exceptions.py

@@ -144,6 +144,10 @@ class ZabbixAPIResponseParsingError(ZabbixAPIRequestError):
     """Zabbix API request error."""
 
 
+class ZabbixAPISessionExpired(ZabbixAPIRequestError):
+    """Zabbix API session expired."""
+
+
 class ZabbixAPICallError(ZabbixAPIException):
     """Zabbix API request error."""
 

zabbix_cli/pyzabbix/client.py

@@ -42,6 +42,7 @@
 from zabbix_cli.exceptions import ZabbixAPINotAuthorizedError
 from zabbix_cli.exceptions import ZabbixAPIRequestError
 from zabbix_cli.exceptions import ZabbixAPIResponseParsingError
+from zabbix_cli.exceptions import ZabbixAPISessionExpired
 from zabbix_cli.exceptions import ZabbixAPITokenExpiredError
 from zabbix_cli.exceptions import ZabbixNotFoundError
 from zabbix_cli.pyzabbix import compat
@@ -455,29 +456,55 @@ def do_request(
                 "Zabbix API returned invalid JSON", response=response
             ) from e
 
-        if resp.error is not None:
-            # some errors don't contain 'data': workaround for ZBX-9340
-            if not resp.error.data:
-                resp.error.data = "No data"
-
-            # TODO: refactor this exc type narrowing to some sort of predicate/dict lookup
-            if "API token expired" in resp.error.data:
-                cls = ZabbixAPITokenExpiredError
-                logger.debug(
-                    "API token '%s' has expired.",
-                    f"{self.auth[:8]}...",  # Redact most of the token
-                )
-            elif "Not authorized" in resp.error.data:
-                cls = ZabbixAPINotAuthorizedError
-            else:
-                cls = ZabbixAPIRequestError
-            raise cls(
-                f"Error: {resp.error.message}: {resp.error.data}",
-                api_response=resp,
-                response=response,
-            )
+        self._check_response_errors(resp, response, params)
+
         return resp
 
+    def _check_response_errors(
+        self,
+        resp: ZabbixAPIResponse,
+        response: httpx.Response,
+        params: ParamsType,
+    ) -> None:
+        # Nothing to handlde
+        if not resp.error:
+            return
+
+        # some errors don't contain 'data': workaround for ZBX-9340
+        if not resp.error.data:
+            resp.error.data = "No data"
+
+        msg = f"Error: {resp.error.message} {resp.error.data}"
+
+        to_replace = [
+            (self.auth, "<token>"),
+            (params.get("token", ""), "<token>"),
+            (params.get("password", ""), "<password>"),
+        ]
+        for replace in to_replace:
+            if replace[0]:
+                msg = msg.replace(str(replace), replace[1])
+
+        # TODO: refactor this exc type narrowing to some sort of predicate/dict lookup
+        msgc = msg.casefold()
+        if "api token expired" in msgc:
+            cls = ZabbixAPITokenExpiredError
+            logger.debug(
+                "API token '%s' has expired.",
+                f"{self.auth[:8]}...",  # Redact most of the token
+            )
+        elif "re-login" in msgc:
+            cls = ZabbixAPISessionExpired
+        elif "not authorized" in msgc:
+            cls = ZabbixAPINotAuthorizedError
+        else:
+            cls = ZabbixAPIRequestError
+        raise cls(
+            msg,
+            api_response=resp,
+            response=response,
+        )
+
     def populate_cache(self) -> None:
         """Populates the various caches with data from the Zabbix API."""
         # NOTE: Must be manually invoked. Can we do this in a thread?