More match-statements

helper/helper/__init__.py

@@ -52,20 +52,21 @@ def _handle_incoming(event, recv, error, cmd_queue):
         if not request:
             break
         try:
-            kind = request["kind"]
+            match request["kind"]:
-            if kind == "signal":
+                case "signal":
-                # Cancel signals are handled here, the rest forwarded
+                    # Cancel signals are handled here, the rest forwarded
-                if request["status"] == "cancel":
+                    if request["status"] == "cancel":
-                    event.set()
+                        logger.debug("Got cancel signal!")
-                else:
+                        event.set()
-                    # Ignore other signals
+                    else:
-                    logger.error("Unhandled signal: %r", request)
+                        # Ignore other signals
-            elif kind == "command":
+                        logger.error("Unhandled signal: %r", request)
-                cmd_queue.join()  # Wait for existing command to complete
+                case "command":
-                event.clear()  # Reset event for next command
+                    cmd_queue.join()  # Wait for existing command to complete
-                cmd_queue.put(request)
+                    event.clear()  # Reset event for next command
-            else:
+                    cmd_queue.put(request)
-                error("invalid-command", "Unsupported request type")
+                case _:
+                    error("invalid-command", "Unsupported request type")
         except KeyError as e:
             error("invalid-command", str(e))
         except RpcException as e:
@@ -171,7 +172,10 @@ def run_rpc_socket(sock):
     def recv():
         line = b""
         while not line.endswith(b"\n"):
-            chunk = sock.recv(1024)
+            try:
+                chunk = sock.recv(1024)
+            except ConnectionError:
+                return None
             if not chunk:
                 return None
             line += chunk

helper/helper/yubiotp.py

@@ -19,12 +19,12 @@ from yubikit.core.otp import modhex_encode, modhex_decode
 from yubikit.yubiotp import (
     YubiOtpSession,
     SLOT,
+    SlotConfiguration,
     UpdateConfiguration,
     HmacSha1SlotConfiguration,
     HotpSlotConfiguration,
     StaticPasswordSlotConfiguration,
     YubiOtpSlotConfiguration,
-    StaticTicketSlotConfiguration,
 )
 from ykman.otp import generate_static_pw, format_csv
 from yubikit.oath import parse_b32_key
@@ -108,15 +108,6 @@ class YubiOtpNode(RpcNode):
         )
 
 
-_CONFIG_TYPES = dict(
-    hmac_sha1=HmacSha1SlotConfiguration,
-    hotp=HotpSlotConfiguration,
-    static_password=StaticPasswordSlotConfiguration,
-    yubiotp=YubiOtpSlotConfiguration,
-    static_ticket=StaticTicketSlotConfiguration,
-)
-
-
 class SlotNode(RpcNode):
     def __init__(self, session, slot):
         super().__init__()
@@ -169,7 +160,8 @@ class SlotNode(RpcNode):
         )
         return dict(response=response)
 
-    def _apply_options(self, config, options):
+    @staticmethod
+    def _apply_options(config, options) -> None:
         for option in (
             "serial_api_visible",
             "serial_usb_visible",
@@ -199,31 +191,29 @@ class SlotNode(RpcNode):
             token_id, *args = options.pop("token_id")
             config.token_id(bytes.fromhex(token_id), *args)
 
-        return config
+    @staticmethod
-
+    def _get_config(cfg_type: str, **kwargs) -> SlotConfiguration:
-    def _get_config(self, type, **kwargs):
+        match cfg_type:
-        config = None
+            case "hmac_sha1":
-
+                return HmacSha1SlotConfiguration(bytes.fromhex(kwargs["key"]))
-        if type in _CONFIG_TYPES:
+            case "hotp":
-            if type == "hmac_sha1":
+                return HotpSlotConfiguration(parse_b32_key(kwargs["key"]))
-                config = _CONFIG_TYPES[type](bytes.fromhex(kwargs["key"]))
+            case "static_password":
-            elif type == "hotp":
+                return StaticPasswordSlotConfiguration(
-                config = _CONFIG_TYPES[type](parse_b32_key(kwargs["key"]))
-            elif type == "static_password":
-                config = _CONFIG_TYPES[type](
                     encode(
                         kwargs["password"], KEYBOARD_LAYOUT[kwargs["keyboard_layout"]]
                     )
                 )
-            elif type == "yubiotp":
+            case "yubiotp":
-                config = _CONFIG_TYPES[type](
+                return YubiOtpSlotConfiguration(
                     fixed=modhex_decode(kwargs["public_id"]),
                     uid=bytes.fromhex(kwargs["private_id"]),
                     key=bytes.fromhex(kwargs["key"]),
                 )
-            else:
+            case unsupported:
-                raise ValueError("No supported configuration type provided.")
+                raise ValueError(
-        return config
+                    f"Unsupported configuration type provided: {unsupported}"
+                )
 
     @action
     def put(
@@ -252,7 +242,7 @@ class SlotNode(RpcNode):
         params,
         acc_code: str | None = None,
         curr_acc_code: str | None = None,
-        **kwargs
+        **kwargs,
     ):
         config = UpdateConfiguration()
         self._apply_options(config, kwargs)