Merge "Implement validate_pkcs10_data"

barbican/common/exception.py

@@ -97,7 +97,7 @@ class InvalidCMCData(BarbicanException):
 
 
 class InvalidPKCS10Data(BarbicanException):
-    message = u._("Invalid PKCS10 Data")
+    message = u._("Invalid PKCS10 Data: %(reason)")
 
 
 class InvalidCertificateRequestType(BarbicanException):

barbican/common/validators.py

@@ -18,6 +18,7 @@ import base64
 
 import jsonschema as schema
 import ldap
+from OpenSSL import crypto
 from oslo_config import cfg
 import six
 
@@ -440,14 +441,24 @@ class TypeOrderValidator(ValidatorBase):
         pass
 
     def _validate_pkcs10_data(self, request_data):
-        """Confirm that the request_data is valid PKCS#10."""
-        """
-        TODO(alee-3) complete this function
+        """Confirm that the request_data is valid PKCS#10.
 
         Parse data into the ASN.1 structure defined by PKCS10.
         If parsing fails, raise InvalidPKCS10Data
         """
-        pass
+        try:
+            csr = crypto.load_certificate_request(crypto.FILETYPE_PEM,
+                                                  request_data)
+        except Exception:
+            reason = u._("Bad format")
+            raise exception.InvalidPKCS10Data(reason=reason)
+
+        try:
+            pubkey = csr.get_pubkey()
+            csr.verify(pubkey)
+        except Exception:
+            reason = u._("Signing key incorrect")
+            raise exception.InvalidPKCS10Data(reason=reason)
 
     def _validate_full_cmc_data(self, request_data):
         """Confirm that request_data is valid Full CMC data."""

barbican/tests/certificate_utils.py

@@ -0,0 +1,65 @@
+# Copyright (c) 2015 Cisco Systems
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+# implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from OpenSSL import crypto
+
+
+def create_key_pair(type, bits):
+    key_pair = crypto.PKey()
+    key_pair.generate_key(type, bits)
+    return key_pair
+
+
+def create_good_csr():
+    """For testing, generate a CSR that will pass validation."""
+    key_pair = create_key_pair(crypto.TYPE_RSA, 1024)
+    csr = crypto.X509Req()
+    subject = csr.get_subject()
+    setattr(subject, "CN", "host.example.net")
+    csr.set_pubkey(key_pair)
+    csr.sign(key_pair, "md5")
+    pem = crypto.dump_certificate_request(crypto.FILETYPE_PEM, csr)
+    return pem
+
+
+def create_csr_that_has_not_been_signed():
+    """For testing, generate a CSR that has not been signed."""
+    key_pair = create_key_pair(crypto.TYPE_RSA, 1024)
+    csr = crypto.X509Req()
+    subject = csr.get_subject()
+    setattr(subject, "CN", "host.example.net")
+    csr.set_pubkey(key_pair)
+    pem = crypto.dump_certificate_request(crypto.FILETYPE_PEM, csr)
+    return pem
+
+
+def create_csr_signed_with_wrong_key():
+    """For testing, generate a CSR that has been signed by the wrong key."""
+    key_pair1 = create_key_pair(crypto.TYPE_RSA, 1024)
+    key_pair2 = create_key_pair(crypto.TYPE_RSA, 1024)
+    csr = crypto.X509Req()
+    subject = csr.get_subject()
+    setattr(subject, "CN", "host.example.net")
+    # set public key from key pair 1
+    csr.set_pubkey(key_pair1)
+    # sign with public key from key pair 2
+    csr.sign(key_pair2, "md5")
+    pem = crypto.dump_certificate_request(crypto.FILETYPE_PEM, csr)
+    return pem
+
+
+def create_bad_csr():
+    """For testing, generate a CSR that will not parse."""
+    return "Bad PKCS10 Data"

barbican/tests/common/test_validators.py

@@ -20,9 +20,9 @@ import testtools
 
 from barbican.common import exception as excep
 from barbican.common import validators
+from barbican.tests import certificate_utils as certs
 from barbican.tests import utils
 
-VALID_PKCS10 = "valid PKCS10"
 VALID_EXTENSIONS = "valid extensions"
 VALID_FULL_CMC = "valid CMC"
 
@@ -1069,7 +1069,7 @@ class WhenTestingSimpleCMCOrderValidator(utils.BaseTestCase):
         super(WhenTestingSimpleCMCOrderValidator, self).setUp()
         self.type = 'certificate'
         self.meta = {'request_type': 'simple-cmc',
-                     'request_data': VALID_PKCS10,
+                     'request_data': certs.create_good_csr(),
                      'requestor_name': 'Barbican User',
                      'requestor_email': 'barbican_user@example.com',
                      'requestor_phone': '555-1212'}
@@ -1103,9 +1103,22 @@ class WhenTestingSimpleCMCOrderValidator(utils.BaseTestCase):
                           self.validator.validate,
                           self.order_req)
 
-    @testtools.skip("Not yet implemented")
     def test_should_raise_with_bad_pkcs10_data(self):
-        self.meta['request_data'] = 'Bad PKCS#10 Data'
+        self.meta['request_data'] = certs.create_bad_csr()
+        self._set_order()
+        self.assertRaises(excep.InvalidPKCS10Data,
+                          self.validator.validate,
+                          self.order_req)
+
+    def test_should_raise_with_signed_wrong_key_pkcs10_data(self):
+        self.meta['request_data'] = certs.create_csr_signed_with_wrong_key()
+        self._set_order()
+        self.assertRaises(excep.InvalidPKCS10Data,
+                          self.validator.validate,
+                          self.order_req)
+
+    def test_should_raise_with_unsigned_pkcs10_data(self):
+        self.meta['request_data'] = certs.create_csr_that_has_not_been_signed()
         self._set_order()
         self.assertRaises(excep.InvalidPKCS10Data,
                           self.validator.validate,