Fix broken unit test.

Turns out that the one remaining old PKCS #11 unit test we weren't
passing was a broken test: code was correctly rejecting CKA_ID
conflicts.  Rewrote test, and added test setup code to use separate
client and server keystores when using the ks_mmap keystore driver.

1 files changed, 54 insertions, 25 deletions

diff --git a/unit_tests.py b/unit_tests.py
index b3c612e..62de2cf 100644
--- a/unit_tests.py
+++ b/unit_tests.py
@@ -25,9 +25,11 @@ def parse_arguments(argv = ()):
     parser.add_argument("--slot",       default = 0, type = int,                        help = "slot number")
     parser.add_argument("--libpkcs11",  default = "./libpkcs11.so",                     help = "PKCS #11 library")
     parser.add_argument("--p11util",    default = "./p11util",                          help = "p11util binary")
-    parser.add_argument("--dbname",     default = "unit_tests.db",                      help = "SQLite3 database")
     parser.add_argument("--server",     default = "../libhal/tests/test-rpc_server",    help = "RPC server binary")
     parser.add_argument("--all-tests",  action = "store_true",                          help = "enable tests usually skipped")
+    parser.add_argument("--sql-file",   default = "unit_tests.db",                      help = "SQLite3 database")
+    parser.add_argument("--ks-client",  default = "unit_tests.ks-client",               help = "client keystore (ks_mmap only)")
+    parser.add_argument("--ks-server",  default = "unit_tests.ks-server",               help = "server keystore (ks_mmap only)")
     return parser.parse_args(argv)
 
 args = parse_arguments()
@@ -38,19 +40,29 @@ rpc  = None
 def setUpModule():
     from subprocess import Popen, PIPE
     from os import unlink, environ, geteuid
-    from os.path import abspath, isfile
+    from os.path import abspath, isfile, expanduser
     global p11
     global rpc
 
+    def new_file(fn):
+        fn = abspath(fn)
+        if isfile(fn):
+            unlink(fn)
+        return fn
+
+    environ["PKCS11_DATABASE"]   = new_file(args.sql_file)
+    environ["CRYPTECH_KEYSTORE"] = new_file(args.ks_client)
+    server_keystore              = new_file(args.ks_server)
+
     if isfile(args.server):
-        if not args.quiet:
-            print "Starting RPC server", args.server
         cmd = [args.server]
         if geteuid() != 0:
             cmd.insert(0, "sudo")
         if not args.quiet:
-            print "Running", " ".join(cmd)
-        rpc = Popen(cmd)
+            print "Starting RPC server:", " ".join(cmd)
+        rpc = Popen(cmd,
+                    env = dict(environ,
+                               CRYPTECH_KEYSTORE = server_keystore))
 
     if not args.quiet:
         print "Setting PINs"
@@ -58,13 +70,6 @@ def setUpModule():
         "{args.so_pin}\n{args.user_pin}\n".format(args = args))
 
     if not args.quiet:
-        print "Configuring database", args.dbname
-    db = abspath(args.dbname)
-    if isfile(db):
-        unlink(db)
-    environ["PKCS11_DATABASE"] = db
-
-    if not args.quiet:
         print "Loading PKCS #11 library", args.libpkcs11
     p11 = PKCS11(args.libpkcs11)
 
@@ -74,7 +79,18 @@ def setUpModule():
 
 def tearDownModule():
     from os import unlink, geteuid
-    unlink(args.dbname)
+    try:
+        unlink(args.sql_file)
+    except:
+        pass
+    try:
+        unlink(args.ks_client)
+    except:
+        pass
+    try:
+        unlink(args.ks_server)
+    except:
+        pass
     global rpc
     if rpc is not None:
         if geteuid() == 0:
@@ -206,9 +222,8 @@ class TestKeys(unittest.TestCase):
         p11.C_CloseAllSessions(args.slot)
         del self.session
 
-    def assertIsKeypair(self, public_handle, private_handle = None):
-        if isinstance(public_handle, tuple) and private_handle is None:
-            public_handle, private_handle = public_handle
+    def assertIsKeypair(self, *keypair):
+        public_handle, private_handle = keypair[0] if isinstance(keypair[0], tuple) else keypair
         self.assertEqual(p11.C_GetAttributeValue(self.session, public_handle,  CKA_CLASS), {CKA_CLASS: CKO_PUBLIC_KEY})
         self.assertEqual(p11.C_GetAttributeValue(self.session, private_handle, CKA_CLASS), {CKA_CLASS: CKO_PRIVATE_KEY})
 
@@ -221,16 +236,30 @@ class TestKeys(unittest.TestCase):
           p11.C_GenerateKeyPair(self.session, CKM_EC_KEY_PAIR_GEN, CKA_TOKEN = True,
                                 CKA_ID = "EC-P256", CKA_EC_PARAMS = self.oid_p256,
                                 CKA_SIGN = True, CKA_VERIFY = True))
+        with self.assertRaises(CKR_Exception):
+            self.assertIsKeypair(
+                p11.C_GenerateKeyPair(self.session, CKM_EC_KEY_PAIR_GEN,
+                                      public_CKA_TOKEN = False, private_CKA_TOKEN = True,
+                                      CKA_ID = "EC-P256", CKA_EC_PARAMS = self.oid_p256,
+                                      CKA_SIGN = True, CKA_VERIFY = True))
+        with self.assertRaises(CKR_Exception):
+            self.assertIsKeypair(
+                p11.C_GenerateKeyPair(self.session, CKM_EC_KEY_PAIR_GEN,
+                                      public_CKA_TOKEN = True, private_CKA_TOKEN = False,
+                                      CKA_ID = "EC-P256", CKA_EC_PARAMS = self.oid_p256,
+                                      CKA_SIGN = True, CKA_VERIFY = True))
+        for handle in p11.FindObjects(self.session):
+            p11.C_DestroyObject(self.session, handle)
         self.assertIsKeypair(
-          p11.C_GenerateKeyPair(self.session, CKM_EC_KEY_PAIR_GEN,
-                                public_CKA_TOKEN = False, private_CKA_TOKEN = True,
-                                CKA_ID = "EC-P256", CKA_EC_PARAMS = self.oid_p256,
-                                CKA_SIGN = True, CKA_VERIFY = True))
+            p11.C_GenerateKeyPair(self.session, CKM_EC_KEY_PAIR_GEN,
+                                  public_CKA_TOKEN = False, private_CKA_TOKEN = True,
+                                  CKA_ID = "EC-P256", CKA_EC_PARAMS = self.oid_p256,
+                                  CKA_SIGN = True, CKA_VERIFY = True))
         self.assertIsKeypair(
-          p11.C_GenerateKeyPair(self.session, CKM_EC_KEY_PAIR_GEN,
-                                public_CKA_TOKEN = True, private_CKA_TOKEN = False,
-                                CKA_ID = "EC-P256", CKA_EC_PARAMS = self.oid_p256,
-                                CKA_SIGN = True, CKA_VERIFY = True))
+            p11.C_GenerateKeyPair(self.session, CKM_EC_KEY_PAIR_GEN,
+                                  public_CKA_TOKEN = True, private_CKA_TOKEN = False,
+                                  CKA_ID = "EC-P256", CKA_EC_PARAMS = self.oid_p256,
+                                  CKA_SIGN = True, CKA_VERIFY = True))
 
     def test_gen_sign_verify_ecdsa_p256_sha256(self):
         public_key, private_key = p11.C_GenerateKeyPair(self.session, CKM_EC_KEY_PAIR_GEN,