From 9bbd41c84d5dc4cc61adeeae67a8db8e4bd06822 Mon Sep 17 00:00:00 2001 From: Thomas AUTRET Date: Wed, 24 Jun 2026 14:38:35 +0200 Subject: [PATCH] Add multithread feature --- pycryptoki/session_management.py | 4 +- tests/functional/test_threading.py | 85 ++++++++++++++++++++++++++++++ tests/unittests/test_threading.py | 60 +++++++++++++++++++++ 3 files changed, 148 insertions(+), 1 deletion(-) create mode 100644 tests/functional/test_threading.py create mode 100644 tests/unittests/test_threading.py diff --git a/pycryptoki/session_management.py b/pycryptoki/session_management.py index fd858f8..cbc7c88 100755 --- a/pycryptoki/session_management.py +++ b/pycryptoki/session_management.py @@ -56,7 +56,9 @@ LOG = logging.getLogger(__name__) -def c_initialize(flags=None, init_struct=None): +from .defines import CKF_OS_LOCKING_OK + +def c_initialize(flags=CKF_OS_LOCKING_OK, init_struct=None): """Initializes current process for use with PKCS11. Some sample flags: diff --git a/tests/functional/test_threading.py b/tests/functional/test_threading.py new file mode 100644 index 0000000..a321b0b --- /dev/null +++ b/tests/functional/test_threading.py @@ -0,0 +1,85 @@ +import threading +import time +import random +import uuid +from pycryptoki.session_management import c_open_session_ex, c_close_session_ex +from pycryptoki.key_generator import c_generate_key_ex +from pycryptoki.session_management import c_get_session_info_ex, c_get_token_info_ex + +class TestThreading: + """ + Test suite to verify that multi-threaded operations do not raise CKR_OPERATION_ACTIVE + or fail under concurrency, thanks to the CKF_OS_LOCKING_OK flag passed during C_Initialize. + """ + + def test_multithreaded_sessions(self, pytestconfig, auth_session, hsm_configured): + """ + Verify that multiple threads can simultaneously open sessions and perform operations + without hitting a concurrency error. + + Note: auth_session from pytest fixture ensures c_initialize is already called. + """ + import pytest + + # auth_session fixture returns the session handle. + # But we need the slot number. The tests in pycryptoki generally use the global pytest config. + # Wait, the auth_session fixture in conftest.py usually logs in, so the token is already logged into + # for this process. We can use c_get_session_info to get the slot from the auth_session handle. + from pycryptoki.session_management import c_get_session_info_ex, login_ex + session_info = c_get_session_info_ex(auth_session) + slot = session_info["slotID"] + + global_password = pytestconfig.getoption("--password") or "1234" + + def thread_task(slot, password, results, index): + try: + print(f"\n[Thread {index}] Starting... opening session") + + # Sleep a tiny random amount to ensure threads interleave + time.sleep(random.uniform(0.01, 0.1)) + + # Attempt to open a session concurrently + h_session = c_open_session_ex(slot) + print(f"[Thread {index}] Session opened: {h_session}") + + # Sleep again to allow other threads to open sessions + time.sleep(random.uniform(0.01, 0.1)) + + print(f"[Thread {index}] Fetching token and session info on {h_session}") + # Interleaved reads from the token + for _ in range(5): + session_info = c_get_session_info_ex(h_session) + token_info = c_get_token_info_ex(slot) + time.sleep(random.uniform(0.01, 0.05)) + print(f"[Thread {index}] Fetched info successfully: session state {session_info['state']}") + + # Close the session + print(f"[Thread {index}] Closing session {h_session}") + c_close_session_ex(h_session) + print(f"[Thread {index}] Finished successfully!\n") + results[index] = True + except Exception as e: + print(f"[Thread {index}] FAILED with exception: {e}\n") + results[index] = e + + + num_threads = 5 + threads = [] + results = [False] * num_threads + + print("\n--- Starting Multithreaded Test ---") + for i in range(num_threads): + t = threading.Thread(target=thread_task, args=(slot, global_password, results, i)) + threads.append(t) + t.start() + + for t in threads: + t.join() + + print("--- Multithreaded Test Finished ---\\n") + + # Check if any thread raised an exception (e.g. CKR_OPERATION_ACTIVE) + errors = [r for r in results if isinstance(r, Exception)] + + # We assert no exceptions were raised + assert not errors, f"Exceptions occurred during multithreaded test: {errors}" diff --git a/tests/unittests/test_threading.py b/tests/unittests/test_threading.py new file mode 100644 index 0000000..1b2f4b8 --- /dev/null +++ b/tests/unittests/test_threading.py @@ -0,0 +1,60 @@ +import mock +from pycryptoki.session_management import c_initialize +from pycryptoki.defines import CKF_OS_LOCKING_OK +from pycryptoki.cryptoki import CK_C_INITIALIZE_ARGS + +class TestInitializationThreading: + """ + Test suite to verify that c_initialize correctly passes the CKF_OS_LOCKING_OK flag + to support multithreaded applications. + """ + + @mock.patch('pycryptoki.session_management.C_Initialize') + def test_c_initialize_default_flags(self, mock_c_initialize): + """ + Verify that c_initialize defaults to CKF_OS_LOCKING_OK. + """ + mock_c_initialize.return_value = 0 + c_initialize() + + # Verify C_Initialize was called + assert mock_c_initialize.called + + # Extract the argument passed to C_Initialize + args, kwargs = mock_c_initialize.call_args + init_struct_ptr = args[0] + + from ctypes import cast, POINTER + + # init_struct_ptr is a pointer to CK_C_INITIALIZE_ARGS (cast to c_void_p) + # we can cast it back to verify flags + init_struct = cast(init_struct_ptr, POINTER(CK_C_INITIALIZE_ARGS)) + assert init_struct.contents.flags == CKF_OS_LOCKING_OK + + @mock.patch('pycryptoki.session_management.C_Initialize') + def test_c_initialize_custom_flags(self, mock_c_initialize): + """ + Verify that c_initialize still respects user-provided flags. + """ + mock_c_initialize.return_value = 0 + + # Pass a different flag, e.g. 0 to mimic default/no OS locking ok + c_initialize(flags=0) + + assert mock_c_initialize.called + + # Extract the argument passed to C_Initialize + args, kwargs = mock_c_initialize.call_args + init_struct_ptr = args[0] + + # If flags=0 is passed, it should either be None if struct isn't created, or 0 if it is created + # But wait, looking at c_initialize implementation, if flags=0, it might be evaluated as false + # let's test a non-zero flag instead. + c_initialize(flags=0x00000001) + + from ctypes import cast, POINTER + + args, kwargs = mock_c_initialize.call_args + init_struct_ptr = args[0] + init_struct = cast(init_struct_ptr, POINTER(CK_C_INITIALIZE_ARGS)) + assert init_struct.contents.flags == 0x00000001