Add PIN management functionality

Additional minor change:
use chardet to detect character encoding of bytestrings and fallback
replacement to prevent utf-8 decoding issues with specific certificates

Author: Dominikus Gierlach

pkcs11/_pkcs11.pyx

@@ -1148,6 +1148,96 @@ cdef class Session(HasFuncList, types.Session):
             op.ingest_chunks(data)
             return op.finish()
 
+    def set_pin(self, old_pin, new_pin):
+        cdef CK_ULONG old_pin_length
+        cdef CK_ULONG new_pin_length
+        cdef CK_OBJECT_HANDLE handle = self.handle
+        cdef CK_UTF8CHAR *old_pin_data
+        cdef CK_UTF8CHAR *new_pin_data
+        cdef CK_RV retval
+
+        if old_pin is None or new_pin is None:
+            raise ArgumentsBad("Set `user_pin`")
+
+        pin_old = old_pin.encode('utf-8')
+        pin_new = new_pin.encode('utf-8')
+
+        if pin_old and pin_new:
+            old_pin_data = pin_old
+            new_pin_data = pin_new
+            old_pin_length = len(pin_old)
+            new_pin_length = len(pin_new)
+
+            with nogil:
+                retval = self.funclist.C_SetPIN(handle, old_pin_data, old_pin_length, new_pin_data, new_pin_length)
+            assertRV(retval)
+            return True
+
+        return False
+
+    def init_pin(self, user_pin):
+        cdef CK_OBJECT_HANDLE handle = self.handle
+        cdef CK_UTF8CHAR *pin_data
+        cdef CK_ULONG pin_length
+        cdef CK_RV retval
+
+        if user_pin is None:
+            raise ArgumentsBad("Set `user_pin`")
+
+        pin = user_pin.encode('utf-8')
+
+        if pin:
+            pin_data = pin
+            pin_length = len(pin)
+
+            with nogil:
+                retval = self.funclist.C_InitPIN(handle, pin_data, pin_length)
+
+            assertRV(retval)
+            return True
+
+        return False
+
+    def logout(self):
+        cdef CK_OBJECT_HANDLE handle = self.handle
+        cdef CK_RV retval
+
+        with nogil:
+            retval = self.funclist.C_Logout(handle)
+
+        assertRV(retval)
+        return True
+
+    def login(self, user_pin, user_type=None):
+        cdef CK_OBJECT_HANDLE handle = self.handle
+        cdef CK_USER_TYPE final_user_type
+        cdef CK_ULONG pin_len
+        cdef CK_UTF8CHAR *pin_data
+        cdef CK_RV retval
+
+        pin = user_pin.encode("utf-8")
+
+        if pin is not None:
+            final_user_type = user_type if user_type is not None else CKU_USER
+            pin_data = pin
+            pin_len = len(pin)
+
+            print("Attempting login as:", final_user_type)
+
+            with nogil:
+                retval = self.funclist.C_Login(handle, final_user_type, pin_data, pin_len)
+
+            print("Login successful as:", final_user_type)
+            assertRV(retval)
+            return True
+        else:
+            pin_data = NULL
+            pin_len = 0
+            final_user_type = UserType.NOBODY
+
+        print("Logged in as:", final_user_type)
+        return False
+
 
 cdef class ObjectHandleWrapper(HasFuncList):
     """

pkcs11/types.py

@@ -7,6 +7,8 @@
 from binascii import hexlify
 from functools import cached_property
 
+import chardet
+
 from pkcs11 import CancelStrategy
 from pkcs11.constants import (
     Attribute,
@@ -41,8 +43,17 @@ def __hash__(self):
 
 
 def _CK_UTF8CHAR_to_str(data):
-    """Convert CK_UTF8CHAR to string."""
-    return data.rstrip(b"\0").decode("utf-8").rstrip()
+    """Convert _CK_UTF8CHAR_to_str to string using chardet for encoding detection."""
+    cleaned_data = data.rstrip(b"\0")
+
+    detected_encoding = chardet.detect(cleaned_data)
+    encoding = detected_encoding["encoding"] or "utf-8"
+
+    try:
+        return cleaned_data.decode(encoding).rstrip()
+    except UnicodeDecodeError:
+        # If decoding fails, try utf-8 with default error replacement character as a last resort
+        return cleaned_data.decode("utf-8", errors="replace").rstrip()
 
 
 def _CK_VERSION_to_tuple(data):
@@ -571,6 +582,18 @@ def digest(self, data, **kwargs):
 
         return self._digest_generator(data, **kwargs)
 
+    def set_pin(self, old_pin, new_pin):
+        raise NotImplementedError()
+
+    def init_pin(self, user_pin):
+        raise NotImplementedError()
+
+    def login(self, user_pin, user_type=None):
+        raise NotImplementedError()
+
+    def logout(self):
+        raise NotImplementedError()
+
     def _digest(self, data, mechanism=None, mechanism_param=None):
         raise NotImplementedError()
 

pyproject.toml

@@ -23,7 +23,10 @@ classifiers = [
     "Programming Language :: Python :: 3.13",
     "Topic :: Security :: Cryptography",
 ]
-dependencies = ["asn1crypto>=1.5.1"]
+dependencies = [
+    "asn1crypto>=1.5.1",
+    "chardet>=5.2.0",
+]
 license = "MIT"
 requires-python = ">=3.9"
 dynamic = ["version"]
@@ -119,4 +122,4 @@ dev = [
     { include-group = "release" },
 ]
 
-[tool.setuptools_scm]
+[tool.setuptools_scm]

tests/test_sessions.py

@@ -291,3 +291,20 @@ def _handler(self, key):
             bool_read = key[Attribute.ID]
             self.assertIsInstance(bool_read, bool)
             self.assertFalse(bool_read, False)
+
+    @Only.softhsm2
+    def test_set_pin(self):
+        old_token_pin = TOKEN_PIN
+        new_token_pin = f"{TOKEN_PIN}56"
+
+        with self.token.open(rw=True, user_pin=old_token_pin) as session:
+            self.assertTrue(session.set_pin(old_token_pin, new_token_pin))
+
+        with self.token.open(user_pin=new_token_pin) as session:
+            self.assertIsInstance(session, pkcs11.Session)
+
+        with self.token.open(rw=True, user_pin=new_token_pin) as session:
+            self.assertTrue(session.set_pin(new_token_pin, old_token_pin))
+
+        with self.token.open(user_pin=old_token_pin) as session:
+            self.assertIsInstance(session, pkcs11.Session)