Merge pull request #1109 from Spinbazz/feat/hsm_login_before_import_pubkey

feat: allow hsm login before import pubkey (pubkey may be inaccessibl…

Author: jku

securesystemslib/signer/_hsm_signer.py

@@ -74,6 +74,10 @@ def PYKCS11LIB():  # type: ignore[no-untyped-def] # noqa: N802
     return _PYKCS11LIB
 
 
+class KeyNotFoundError(ValueError):
+    pass
+
+
 class HSMSigner(Signer):
     """Hardware Security Module (HSM) Signer.
 
@@ -220,7 +224,7 @@ def _find_key(
             ]
         )
         if not keys:
-            raise ValueError(f"could not find {_KEY_TYPE_ECDSA} key for {keyid}")
+            raise KeyNotFoundError(f"could not find {_KEY_TYPE_ECDSA} key for {keyid}")
 
         if len(keys) > 1:
             raise ValueError(f"found more than one {_KEY_TYPE_ECDSA} key for {keyid}")
@@ -266,6 +270,7 @@ def import_(
         cls,
         hsm_keyid: int | None = None,
         token_filter: dict[str, str] | None = None,
+        pin_handler: SecretsHandler | None = None,
     ) -> tuple[str, SSlibKey]:
         """Import public key and signer details from HSM.
 
@@ -305,9 +310,15 @@ def import_(
             token_filter = cls._build_token_filter()
 
         uri = f"{cls.SCHEME}:{hsm_keyid}?{parse.urlencode(token_filter)}"
-
         with cls._get_session(token_filter) as session:
-            params, point = cls._find_key_values(session, hsm_keyid)
+            try:
+                params, point = cls._find_key_values(session, hsm_keyid)
+            except KeyNotFoundError as e:
+                if pin_handler is not None:
+                    session.login(pin_handler(cls.SECRETS_HANDLER_MSG))
+                    params, point = cls._find_key_values(session, hsm_keyid)
+                else:
+                    raise e
 
         if params.chosen.native not in _CURVE_NAMES:
             raise ValueError(

tests/test_hsm_signer.py

@@ -4,6 +4,7 @@
 import shutil
 import tempfile
 import unittest
+from unittest.mock import Mock
 
 from asn1crypto.keys import (
     ECDomainParameters,
@@ -31,18 +32,22 @@ class TestHSM(unittest.TestCase):
     hsm_keyid = 1
     hsm_keyid_default = 2
     hsm_keyid_odd = 258
+    hsm_keyid_priv = 512
     hsm_user_pin = "123456"
 
     @staticmethod
-    def _generate_key_pair(session, keyid, curve):
+    def _generate_key_pair(session, keyid, curve, is_pubkey_private=False):
         "Create ecdsa key pair on hsm"
         params = ECDomainParameters(name="named", value=NamedCurve(curve.name)).dump()
 
         cka_id = list(keyid.to_bytes((keyid.bit_length() + 7) // 8 or 1, "big"))
 
         public_template = [
             (PyKCS11.CKA_CLASS, PyKCS11.CKO_PUBLIC_KEY),
-            (PyKCS11.CKA_PRIVATE, PyKCS11.CK_FALSE),
+            (
+                PyKCS11.CKA_PRIVATE,
+                PyKCS11.CK_TRUE if is_pubkey_private else PyKCS11.CK_FALSE,
+            ),
             (PyKCS11.CKA_TOKEN, PyKCS11.CK_TRUE),
             (PyKCS11.CKA_ENCRYPT, PyKCS11.CK_FALSE),
             (PyKCS11.CKA_VERIFY, PyKCS11.CK_TRUE),
@@ -104,6 +109,9 @@ def setUpClass(cls):
         cls._generate_key_pair(session, cls.hsm_keyid, SECP256R1)
         cls._generate_key_pair(session, cls.hsm_keyid_default, SECP384R1)
         cls._generate_key_pair(session, cls.hsm_keyid_odd, SECP256R1)
+        cls._generate_key_pair(
+            session, cls.hsm_keyid_priv, SECP256R1, is_pubkey_private=True
+        )
 
         session.logout()
         session.closeSession()
@@ -147,6 +155,48 @@ def test_hsm_uri(self):
         with self.assertRaises(UnverifiedSignatureError):
             key.verify_signature(sig, b"NOT DATA")
 
+    def test_hsm_private_pubkey(self):
+        """Test PIN"""
+
+        # Test pub key not found without pin
+        with self.assertRaises(Exception) as context:
+            _, _ = HSMSigner.import_(self.hsm_keyid_priv, self.token_filter)
+            self.assertTrue("could not fin" in context.exception)
+
+        # or with bad pin
+        bad_pinh = Mock(return_value="987")
+        with self.assertRaises(Exception) as context:
+            _, _ = HSMSigner.import_(
+                self.hsm_keyid_priv, self.token_filter, pin_handler=bad_pinh
+            )
+            self.assertTrue("could not fin" in context.exception)
+        self.assertTrue(bad_pinh.assert_called_once)
+
+        # And check sign ok cases (good pin or not required pin)
+        for hsm_keyid in [
+            self.hsm_keyid,
+            self.hsm_keyid_default,
+            self.hsm_keyid_odd,
+            self.hsm_keyid_priv,
+        ]:
+            good_pinh = Mock(return_value=self.hsm_user_pin)
+            _, key = HSMSigner.import_(
+                hsm_keyid, self.token_filter, pin_handler=good_pinh
+            )
+            # Pin handler must be called only if public key is private
+            if hsm_keyid == self.hsm_keyid_priv:
+                good_pinh.assert_called_once()
+            else:
+                good_pinh.assert_not_called()
+
+            signer = HSMSigner(hsm_keyid, self.token_filter, key, good_pinh)
+
+            sig = signer.sign(b"DATA")
+            key.verify_signature(sig, b"DATA")
+
+            with self.assertRaises(UnverifiedSignatureError):
+                key.verify_signature(sig, b"NOT DATA")
+
 
 # Run the unit tests.
 if __name__ == "__main__":