Improve the error message for bad pkcs12 bundles

When a user loads a bad pkcs12 bundle or one with a pass phrase into
barbican and then uses it for a TLS-TERMINATED listener, the error
we return the user is misleading[1].
This patch improves the error message to point out that we got the
bundle from barbican, but that it is unreadable and/or protected
with a pass phrase.

[1] Could not retrieve certificate: [ ... ] (HTTP 400)

Conflicts:
    octavia/api/v2/controllers/listener.py

Change-Id: I6ad0349dba62b1141be07bfb0e40171e9f7a91b9
Story: 2006587
Task: 36713
(cherry picked from commit a0f4335c38)
(cherry picked from commit a501714a76)
(cherry picked from commit 08916abd2b)

octavia/api/v2/controllers/listener.py

@@ -142,6 +142,8 @@ class ListenersController(base.BaseController):
             try:
                 cert_parser.load_certificate_data(
                     self.cert_manager, ref, context)
+            except exceptions.UnreadablePKCS12:
+                raise
             except Exception:
                 bad_refs.append(ref)
 

octavia/certificates/common/pkcs12.py

@@ -21,12 +21,16 @@ from cryptography.hazmat.primitives import serialization
 from OpenSSL import crypto
 
 from octavia.certificates.common import cert
+from octavia.common import exceptions
 
 
 class PKCS12Cert(cert.Cert):
     """Representation of a Cert for local storage."""
     def __init__(self, certbag):
-        p12 = crypto.load_pkcs12(certbag)
+        try:
+            p12 = crypto.load_pkcs12(certbag)
+        except crypto.Error as e:
+            raise exceptions.UnreadablePKCS12(error=str(e))
         self.certificate = p12.get_certificate()
         self.intermediates = p12.get_ca_certificates()
         self.private_key = p12.get_privatekey()

octavia/certificates/manager/barbican.py

@@ -112,6 +112,8 @@ class BarbicanCertManager(cert_mgr.CertManager):
         try:
             cert_secret = connection.secrets.get(secret_ref=cert_ref)
             return pkcs12.PKCS12Cert(cert_secret.payload)
+        except exceptions.UnreadablePKCS12:
+            raise
         except Exception:
             # If our get fails, try with the legacy driver.
             # TODO(rm_work): Remove this code when the deprecation cycle for

octavia/common/exceptions.py

@@ -128,6 +128,13 @@ class UnreadableCert(OctaviaException):
     message = _("Could not read X509 from PEM")
 
 
+class UnreadablePKCS12(APIException):
+    msg = _("The PKCS12 bundle is unreadable. Please check the PKCS12 bundle "
+            "validity. In addition, make sure it does not require a pass "
+            "phrase. Error: %(error)s")
+    code = 400
+
+
 class MisMatchedKey(OctaviaException):
     message = _("Key and x509 certificate do not match")
 

octavia/tests/unit/certificates/manager/test_barbican.py

@@ -16,10 +16,12 @@ import uuid
 
 from barbicanclient.v1 import secrets
 import mock
+from OpenSSL import crypto
 
 import octavia.certificates.common.barbican as barbican_common
 import octavia.certificates.common.cert as cert
 import octavia.certificates.manager.barbican as barbican_cert_mgr
+from octavia.common import exceptions
 import octavia.tests.unit.base as base
 import octavia.tests.unit.common.sample_configs.sample_certs as sample
 
@@ -133,6 +135,22 @@ class TestBarbicanManager(base.TestCase):
         self.assertItemsEqual(sample.X509_IMDS_LIST, data.get_intermediates())
         self.assertIsNone(data.get_private_key_passphrase())
 
+    @mock.patch('OpenSSL.crypto.load_pkcs12')
+    def test_get_cert_bad_pkcs12(self, mock_load_pkcs12):
+
+        mock_load_pkcs12.side_effect = [crypto.Error]
+
+        # Mock out the client
+        self.bc.secrets.get.return_value = self.secret
+
+        # Test bad pkcs12 bundle re-raises UnreadablePKCS12
+        self.assertRaises(exceptions.UnreadablePKCS12,
+                          self.cert_manager.get_cert,
+                          context=self.context,
+                          cert_ref=self.secret_ref,
+                          resource_ref=self.secret_ref,
+                          service_name='Octavia')
+
     def test_delete_cert_legacy(self):
         # Attempt to deregister as a consumer
         self.cert_manager.delete_cert(