Merge "Ignore network errors during C_Finalize"

barbican/common/exception.py

@@ -339,6 +339,10 @@ class P11CryptoTokenException(PKCS11Exception):
     message = u._("No token was found in slot %(slot_id)s")
 
 
+class TrustwayProteccioException(PKCS11Exception):
+    message = u._("Trustway Proteccio HSM Error")
+
+
 class MultipleStorePreferredPluginMissing(BarbicanException):
     """Raised when a preferred plugin is missing in service configuration."""
     def __init__(self, store_name):

barbican/plugin/crypto/pkcs11.py

@@ -259,7 +259,9 @@ ERROR_CODES = {
     0x1a0: 'CKR_MUTEX_BAD',
     0x1a1: 'CKR_MUTEX_NOT_LOCKED',
     0x200: 'CKR_FUNCTION_REJECTED',
-    1 << 31: 'CKR_VENDOR_DEFINED'
+    1 << 31: 'CKR_VENDOR_DEFINED',
+    # Trustway Proteccio Codes
+    0x81000071: 'EHOSTUNREACH'
 }
 
 
@@ -857,7 +859,10 @@ class PKCS11(object):
 
     def finalize(self):
         rv = self.lib.C_Finalize(self.ffi.NULL)
-        self._check_error(rv)
+        try:
+            self._check_error(rv)
+        except exception.TrustwayProteccioException:
+            LOG.warning("Trustway Proteccio client failed to finalize.")
 
     def _check_error(self, value):
         if value != CKR_OK and value != CKR_CRYPTOKI_ALREADY_INITIALIZED:
@@ -867,6 +872,10 @@ class PKCS11(object):
             if code == 'CKR_TOKEN_NOT_PRESENT':
                 raise exception.P11CryptoTokenException(slot_id=self.slot_id)
 
+            if code == 'EHOSTUNREACH':
+                raise exception.TrustwayProteccioException(
+                    "Trustway Proteccio Error: {code}".format(code=hex_code))
+
             raise exception.P11CryptoPluginException(u._(
                 "HSM returned response code: {code}").format(code=hex_code))
 

barbican/tests/plugin/crypto/test_pkcs11.py

@@ -456,6 +456,12 @@ class WhenTestingPKCS11(utils.BaseTestCase):
 
         self.assertEqual(1, self.lib.C_Finalize.call_count)
 
+    def test_finalize_ignores_trustway_network_errors(self):
+        self.lib.C_Finalize.return_value = 0x81000071
+        self.pkcs11.finalize()
+
+        self.assertEqual(1, self.lib.C_Finalize.call_count)
+
     def test_check_error(self):
         self.assertIsNone(self.pkcs11._check_error(pkcs11.CKR_OK))