feat(api): validate a certificate's private key

rootfs/api/models/certificate.py

@@ -66,11 +66,18 @@ def get_subj_alt_name(peer_cert):
 
 def validate_certificate(value):
     try:
-        crypto.load_certificate(crypto.FILETYPE_PEM, value)
+        return crypto.load_certificate(crypto.FILETYPE_PEM, value)
     except crypto.Error as e:
         raise ValidationError('Could not load certificate: {}'.format(e))
 
 
+def validate_private_key(value):
+    try:
+        return crypto.load_privatekey(crypto.FILETYPE_PEM, value)
+    except crypto.Error as e:
+        raise ValidationError('Could not load private key: {}'.format(e))
+
+
 class Certificate(AuditedModel):
     """
     Public and private key pair used to secure application traffic at the router.
@@ -79,7 +86,7 @@ class Certificate(AuditedModel):
     name = models.CharField(max_length=253, unique=True, validators=[validate_label])
     # there is no upper limit on the size of an x.509 certificate
     certificate = models.TextField(validators=[validate_certificate])
-    key = models.TextField()
+    key = models.TextField(validators=[validate_private_key])
     # X.509 certificates allow any string of information as the common name.
     common_name = models.TextField(editable=False, unique=False, null=True)
     # A list of DNS records if certificate has SubjectAltName
@@ -106,14 +113,14 @@ def domains(self):
     def __str__(self):
         return self.name
 
-    def _get_certificate(self):
+    def save(self, *args, **kwargs):
         try:
-            return crypto.load_certificate(crypto.FILETYPE_PEM, self.certificate)
-        except crypto.Error as e:
+            certificate = validate_certificate(self.certificate)
+            # NOTE(bacongobbler): we want to load the key here to ensure that it is valid before
+            # saving it to the database.
+            validate_private_key(self.key)
+        except ValidationError as e:
             raise SuspiciousOperation(e)
-
-    def save(self, *args, **kwargs):
-        certificate = self._get_certificate()
         if not self.common_name:
             self.common_name = certificate.get_subject().CN
 

rootfs/api/tests/test_certificate.py

@@ -145,7 +145,8 @@ def test_delete_certificate(self):
             name='random-test-cert',
             owner=self.user,
             common_name='autotest.example.com',
-            certificate=self.cert
+            certificate=self.cert,
+            key=self.key
         )
         url = '/v2/certs/random-test-cert'
         response = self.client.delete(url)
@@ -176,6 +177,31 @@ def test_load_invalid_cert(self):
                 key='i am bad data as well'
             )
 
+    def test_create_invalid_key(self):
+        """Upload a private key that can't be loaded by pyopenssl"""
+        response = self.client.post(
+            self.url,
+            {
+                'name': 'random-test-cert',
+                'certificate': self.cert,
+                'key': 'I am Groot.'
+            }
+        )
+        self.assertEqual(response.status_code, 400, response.data)
+        # match partial since right now the rest is pyopenssl errors
+        self.assertIn('Could not load private key', response.data['key'][0])
+
+    def test_load_invalid_key(self):
+        """Inject a private key that can't be loaded by pyopenssl"""
+
+        with self.assertRaises(SuspiciousOperation):
+            Certificate.objects.create(
+                owner=self.user,
+                name='random-test-cert',
+                certificate=self.cert,
+                key='I am Groot.'
+            )
+
     def test_certs_fetch_limit(self):
         """
         When a user retrieves a certificate, make sure limits work

rootfs/api/tests/test_certificate_use_case_1.py

@@ -144,7 +144,8 @@ def test_delete_certificate(self):
         Certificate.objects.create(
             owner=self.user,
             name=self.name,
-            certificate=self.cert
+            certificate=self.cert,
+            key=self.key
         )
 
         url = '/v2/certs/{}'.format(self.name)

rootfs/api/tests/test_certificate_use_case_2.py

@@ -128,7 +128,8 @@ def test_delete_certificate(self):
                 name=certificate['name'],
                 owner=self.user,
                 common_name=domain,
-                certificate=certificate['cert']
+                certificate=certificate['cert'],
+                key=certificate['key']
             )
             url = '/v2/certs/{}'.format(certificate['name'])
             response = self.client.delete(url)

rootfs/api/tests/test_certificate_use_case_3.py

@@ -130,7 +130,8 @@ def test_delete_certificate(self):
                 name=certificate['name'],
                 owner=self.user,
                 common_name=domain,
-                certificate=certificate['cert']
+                certificate=certificate['cert'],
+                key=certificate['key']
             )
             url = '/v2/certs/{}'.format(certificate['name'])
             response = self.client.delete(url)

rootfs/api/tests/test_certificate_use_case_4.py

@@ -212,7 +212,8 @@ def test_delete_certificate(self):
                 name=certificate['name'],
                 owner=self.user,
                 common_name=domain,
-                certificate=certificate['cert']
+                certificate=certificate['cert'],
+                key=certificate['key']
             )
             url = '/v2/certs/{}'.format(certificate['name'])
             response = self.client.delete(url)

rootfs/api/tests/test_certificate_use_case_5.py

@@ -158,7 +158,8 @@ def test_delete_certificate(self):
                 name=certificate['name'],
                 owner=self.user,
                 common_name=domain,
-                certificate=certificate['cert']
+                certificate=certificate['cert'],
+                key=certificate['key']
             )
 
             # Remove certificate