Unit tests for init, session, and login functions.

4 files changed, 63 insertions, 53 deletions

diff --git a/GNUmakefile b/GNUmakefile
index 89f713c..77daebd 100644
--- a/GNUmakefile
+++ b/GNUmakefile
@@ -39,7 +39,7 @@ LIBHAL_DIR   = ../libhal
 LIBTFM_DIR   = ../thirdparty/libtfm
 SQLITE3_DIR  = ../thirdparty/sqlite3
 
-CFLAGS	:= -g3 -fPIC -Wall -std=c99 -I${LIBHAL_DIR} -I${SQLITE3_DIR}
+CFLAGS	:= -g3 -fPIC -Wall -std=c99 -I${LIBHAL_DIR} -I${SQLITE3_DIR} -DDEBUG_PKCS11=0
 SOFLAGS := -Wl,-Bsymbolic-functions -Wl,-Bsymbolic -Wl,-z,noexecstack
 LIBS	:= ${LIBHAL_DIR}/libhal.a ${LIBTFM_DIR}/libtfm.a ${SQLITE3_DIR}/libsqlite3.a
 
@@ -90,7 +90,7 @@ tags: TAGS
 TAGS: *.[ch]
 	etags $^
 
-test:
+test: all
 	sudo python unit_tests.py
 
 # Kudge for testing, gussy this up if we decide to keep it
diff --git a/pkcs11.c b/pkcs11.c
index 93a5f36..eef1aec 100644
--- a/pkcs11.c
+++ b/pkcs11.c
@@ -1593,6 +1593,10 @@ static CK_RV p11_session_delete(const CK_SESSION_HANDLE session_handle)
   p11_sessions = session->link;
   p11_session_free(session);
 
+  /* Deleting last session also logs us out */
+  if (p11_sessions == NULL)
+    logged_in_as = not_logged_in;
+
  fail:
   sqlite3_finalize(q);
   return rv;
@@ -1621,6 +1625,8 @@ static CK_RV p11_session_delete_all(void)
     p11_session_free(session);
   }
 
+  logged_in_as = not_logged_in;
+
  fail:
   return rv;
 }
@@ -2696,6 +2702,11 @@ CK_RV C_Finalize(CK_VOID_PTR pReserved)
   rv =  mutex_unlock(p11_global_mutex);
   (void) mutex_destroy(p11_global_mutex);
   p11_global_mutex = NULL;
+
+#if USE_POSIX
+  initialized_pid = 0;
+#endif
+
   return rv;
 
  fail:
diff --git a/py11/__init__.py b/py11/__init__.py
index d800ab7..ec51d19 100644
--- a/py11/__init__.py
+++ b/py11/__init__.py
@@ -83,10 +83,10 @@ class PKCS11 (object):
       self._C_Initialize_args = None
       self.so.C_Initialize(None)
     else:
-      create_mutex  = cast(None, CK_CREATEMUTEX)  if create_mutex  is None else CK_CREATEMUTEX(create_mutex)
-      destroy_mutex = cast(None, CK_DESTROYMUTEX) if destroy_mutex is None else CK_DESTROYMUTEX(destroy_mutex)
-      lock_mutex    = cast(None, CK_LOCKMUTEX)    if lock_mutex    is None else CK_LOCKMUTEX(lock_mutex)
-      unlock_mutex  = cast(None, CK_UNLOCKMUTEX)  if unlock_mutex  is None else CK_UNLOCKMUTEX(unlock_mutex)
+      create_mutex  = CK_CREATEMUTEX()  if create_mutex  is None else CK_CREATEMUTEX(create_mutex)
+      destroy_mutex = CK_DESTROYMUTEX() if destroy_mutex is None else CK_DESTROYMUTEX(destroy_mutex)
+      lock_mutex    = CK_LOCKMUTEX()    if lock_mutex    is None else CK_LOCKMUTEX(lock_mutex)
+      unlock_mutex  = CK_UNLOCKMUTEX()  if unlock_mutex  is None else CK_UNLOCKMUTEX(unlock_mutex)
       self._C_Initialize_args = CK_C_INITIALIZE_ARGS(create_mutex, destroy_mutex, lock_mutex, unlock_mutex, flags, None)
       self.so.C_Initialize(cast(byref(self._C_Initialize_args), CK_VOID_PTR))
 
@@ -105,7 +105,8 @@ class PKCS11 (object):
     self.so.C_GetTokenInfo(slot_id, byref(token_info))
     return token_info
 
-  def C_OpenSession(self, slot, flags = CKF_SERIAL_SESSION | CKF_RW_SESSION, application = None, notify = CK_NOTIFY()):
+  def C_OpenSession(self, slot, flags = CKF_RW_SESSION, application = None, notify = CK_NOTIFY()):
+    flags |= CKF_SERIAL_SESSION
     handle = CK_SESSION_HANDLE()
     self.so.C_OpenSession(slot, flags, application, notify, byref(handle))
     return handle.value
diff --git a/unit_tests.py b/unit_tests.py
index a59b731..2e4dc5d 100644
--- a/unit_tests.py
+++ b/unit_tests.py
@@ -8,6 +8,10 @@ from py11.mutex import MutexDB
 
 p11 = None
 
+so_pin    = "fnord"
+user_pin  = "fnord"
+only_slot = 0
+
 class TestInit(unittest.TestCase):
 
   def test_no_lock(self):
@@ -29,63 +33,55 @@ class TestInit(unittest.TestCase):
 
 class TestDevice(unittest.TestCase):
 
-  def test_slots(self):
-    slots = p11.C_GetSlotList()
-    self.assertIsInstance(slots, (list, tuple))
-    self.assertEqual(len(slots), 1)
-    self.assertIsInstance(slots[0], (int, long))
-
-@unittest.skip("This was just an example")
-class Test1(unittest.TestCase):
-
-  @classmethod
-  def setUpClass(cls):
-    pass #print "Class-wide setup"
-
-  @classmethod
-  def tearDownClass(cls):
-    pass #print "Class-wide teardown"
-
-  def setUp(self):
-    pass #print "Per-test setup"
-
-  def tearDown(self):
-    pass #print "Per-test teardown"
-
-  def test_one(self):
-    pass #print "Test one"
-    self.assertTrue(True)
-
-  def test_two(self):
-    pass #print "Test two"
-    self.assertTrue(True)
-
-@unittest.skip("This was also just an example")
-class Test2(unittest.TestCase):
-
   @classmethod
   def setUpClass(cls):
-    pass #print "Class-wide setup"
+    p11.C_Initialize()
 
   @classmethod
   def tearDownClass(cls):
-    pass #print "Class-wide teardown"
-
-  def setUp(self):
-    pass #print "Per-test setup"
+    p11.C_Finalize()
 
   def tearDown(self):
-    pass #print "Per-test teardown"
+    p11.C_CloseAllSessions(only_slot)
 
-  def test_three(self):
-    pass #print "Test three"
-    self.assertTrue(True)
+  def test_slots(self):
+    self.assertEqual(p11.C_GetSlotList(), [only_slot])
+
+  def test_serial_sessions(self):
+    rw_session = p11.C_OpenSession(only_slot, CKF_RW_SESSION | CKF_SERIAL_SESSION)
+    ro_session = p11.C_OpenSession(only_slot, CKF_SERIAL_SESSION)
+
+  def test_parallel_sessions(self):
+    # Cooked API doesn't allow the user to make this mistake, so we
+    # have to use the raw API to test it.
+    from ctypes import byref
+    handle = CK_SESSION_HANDLE()
+    notify = CK_NOTIFY()
+    with self.assertRaises(CKR_SESSION_PARALLEL_NOT_SUPPORTED):
+      p11.so.C_OpenSession(only_slot, CKF_RW_SESSION, None, notify, byref(handle))
+    with self.assertRaises(CKR_SESSION_PARALLEL_NOT_SUPPORTED):
+      p11.so.C_OpenSession(only_slot, 0,              None, notify, byref(handle))
+
+  def test_login_user(self):
+    rw_session = p11.C_OpenSession(only_slot, CKF_RW_SESSION | CKF_SERIAL_SESSION)
+    ro_session = p11.C_OpenSession(only_slot, CKF_SERIAL_SESSION)
+    p11.C_Login(ro_session, CKU_USER, user_pin)
+    p11.C_Logout(ro_session)
+
+  def test_login_so(self):
+    rw_session = p11.C_OpenSession(only_slot, CKF_RW_SESSION | CKF_SERIAL_SESSION)
+    ro_session = p11.C_OpenSession(only_slot, CKF_SERIAL_SESSION)
+    self.assertRaises(CKR_SESSION_READ_ONLY_EXISTS, p11.C_Login, ro_session, CKU_SO, user_pin)
+    p11.C_CloseSession(ro_session)
+    p11.C_Login(rw_session, CKU_SO, user_pin)
+    self.assertRaises(CKR_SESSION_READ_WRITE_SO_EXISTS, p11.C_OpenSession, only_slot, CKF_SERIAL_SESSION)
+    p11.C_Logout(rw_session)
 
 def setUpModule():
   from os import environ
   from os.path import abspath
   global p11
-  p11 = PKCS11()
+  p11 = PKCS11("./libpkcs11.so")
   environ["PKCS11_DATABASE"] = abspath("unit_tests.db")
   delete_db()
   set_pin()
@@ -95,15 +91,17 @@ def tearDownModule():
 
 def delete_db():
   from os import environ, unlink
+  print "Deleting database", environ["PKCS11_DATABASE"]
   try:
     unlink(environ["PKCS11_DATABASE"])
   except OSError, e:
     if e.errno != 2:
       raise
 
-def set_pin(user_pin = "fnord", so_pin = "fnord"):
+def set_pin():
   from subprocess import Popen, PIPE
-  Popen(("./p11util", "-sup"), stdin = PIPE).communicate("fnord\nfnord\n")
+  print "Creating database and setting PINs"
+  Popen(("./p11util", "-sup"), stdin = PIPE).communicate("%s\n%s\n" % (so_pin, user_pin))
 
 if __name__ == "__main__":
   unittest.main(verbosity = 2)